INTERMEDIATE — Lesson 01

Lesson 01 — Functions Done Right

Source files: scripts/daily-content-agent.py and scripts/trendpilot-loop.py

---

The Problem

You have qwen_call() defined in two separate scripts. Both versions have the

same flaw: they return different types depending on what happens.

From scripts/daily-content-agent.py, lines 212–238:

def qwen_call(prompt, model="qwen-plus", temperature=0.7, max_tokens=4000):
    if not DASHSCOPE_API_KEY:
        return {"error": "No DASHSCOPE_API_KEY set"}   # returns a DICT
    # ...
    try:
        # ...
        return data["choices"][0]["message"]["content"]  # returns a STRING
    except Exception as e:
        return {"error": str(e)}                         # returns a DICT

One function, three possible return types. Now look at every place that calls it:

# daily-content-agent.py line 424
article = qwen_call(prompt, ...)
if isinstance(article, dict) and "error" in article:
    print(f"  ERROR Qwen API error: {article['error']}")
    # ... handle error

# trendpilot-loop.py line 1013
result = qwen_call(description_prompt, ...)
if isinstance(result, str) and len(result) > 50:
    qwen_description = result

You're writing isinstance() checks everywhere because you can't trust what

comes back. That's a red flag — it means the function's contract is unclear.

---

Why This Matters

A function's signature is its contract: "give me these inputs, I'll give you

this output." When the output type varies, the caller has to guess. That leads to:

1. Defensive isinstance() checks cluttering your code

2. Bugs when you forget to check (silently use a dict as a string)

3. No way to use type hints to catch it before running

---

The Fix: Raise Exceptions, Don't Return Error Dicts

The Python way to signal failure is to raise an exception, not return a

special error value. Callers that don't care about errors don't have to do

anything — they let the exception bubble up. Callers that do care use try/except.

class QwenError(Exception):
    """Raised when the Qwen API call fails."""
    pass

def qwen_call(prompt, model="qwen-plus", temperature=0.7, max_tokens=4000) -> str:
    """Call Qwen Cloud API. Always returns a string. Raises QwenError on failure."""
    if not DASHSCOPE_API_KEY:
        raise QwenError("No DASHSCOPE_API_KEY set")

    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "max_tokens": max_tokens,
        "stream": False,
    }
    try:
        req = urllib.request.Request(
            QWEN_API_URL,
            data=json.dumps(payload).encode(),
            headers=headers,
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=120) as resp:
            data = json.loads(resp.read())
            return data["choices"][0]["message"]["content"]
    except urllib.error.HTTPError as e:
        raise QwenError(f"HTTP {e.code}: {e.read().decode()[:200]}") from e
    except Exception as e:
        raise QwenError(str(e)) from e

Notice: the return type annotation -> str tells Python (and you) exactly what

comes back. No dicts, no isinstance() checks at the call site.

Now the callers become much cleaner:

# Before: 5 lines of isinstance checking
article = qwen_call(prompt)
if isinstance(article, dict) and "error" in article:
    print(f"  ERROR: {article['error']}")
    file_path.write_text(f"# {keyword}\n\n*Writing...*\n")
    return str(file_path)
file_path.write_text(article)

# After: clear intent, less code
try:
    article = qwen_call(prompt)
    file_path.write_text(article)
except QwenError as e:
    print(f"  ERROR: {e}")
    file_path.write_text(f"# {keyword}\n\n*Writing...*\n")
    return str(file_path)

---

Key Concepts

Return type consistency

A function should return the same type every time. If it sometimes fails, it

should raise an exception — not return a different type.

`-> type` annotations

The -> str in def qwen_call(...) -> str: is a return type hint. Python

doesn't enforce it at runtime, but it documents the contract and tools like

mypy can catch violations.

Custom exceptions

class QwenError(Exception): pass creates a new exception type. This lets

callers catch *your specific error* without catching everything:

try:
    result = qwen_call(prompt)
except QwenError:
    # Only catches Qwen failures, not your own bugs
    handle_qwen_failure()

Compare to catching Exception — that catches everything including your typos,

which hides bugs.

The `from e` in `raise QwenError(...) from e`

This chains exceptions. If someone prints the traceback, they'll see *both* the

original error and your QwenError. You don't lose the original context.

---

Default Arguments

Your functions use default arguments well:

def qwen_call(prompt, model="qwen-plus", temperature=0.7, max_tokens=4000):

Caller only has to pass prompt. The rest have sensible defaults. This is good.

One gotcha: never use a mutable default argument.

# BAD — this list is shared across ALL calls
def pick_keywords(product, day_index, keywords=[]):
    keywords.append(product)  # modifies the shared list!

# GOOD — use None as sentinel, create fresh list inside
def pick_keywords(product, day_index, keywords=None):
    if keywords is None:
        keywords = []

You don't have this bug currently, but it's worth knowing because it's a common

Python trap.

---

Exercises

Exercise 1 — Read the code:

Open scripts/trendpilot-loop.py. Find the qwen_call function (around line

166). Count how many places in the same file check isinstance(result, dict).

Answer: ___

Exercise 2 — Trace a call:

In scripts/daily-content-agent.py, the function expand_keywords() (line 477)

calls qwen_call and then does if isinstance(result, dict) and "error" in result.

If you changed qwen_call to raise QwenError instead, what would the expand_keywords

function look like? Rewrite just that section.

Exercise 3 — Write the fix:

Create a file scripts/qwen_client.py with a single qwen_call() function that:

• Raises `QwenError` on failure
• Always returns a `str` on success
• Has `-> str` return annotation
• Can be imported into both `daily-content-agent.py` and `trendpilot-loop.py`

Then update one caller in daily-content-agent.py to use it.

Exercise 4 — Think about it:

The function write_article() in daily-content-agent.py currently returns a

str (the file path). But what should it return if the Qwen API is down and it

wrote a placeholder? Should it raise? Return None? Return the path anyway?

There's no single right answer — write down your reasoning.

INTERMEDIATE — Lesson 02

Lesson 02 — Error Handling

Source files: scripts/fb-reel-upload.py, scripts/trendpilot-loop.py

---

The Crash Bug in fb-reel-upload.py

This script has a real bug that crashes if you use --page epic. Look at

lines 184–220:

# Upload to Epic Trends Store
if PAGE_TARGET in ["epic", "both"]:
    print("🚀 Uploading to Epic Trends Store...")
    epic_video_id, epic_status, epic_details = upload_video_to_fb(...)
    # ...

# Upload to BrandBoost Studio
if PAGE_TARGET in ["brandboost", "both"]:
    print("🚀 Uploading to BrandBoost Studio...")
    brandboost_video_id, brandboost_status, brandboost_details = upload_video_to_fb(...)

if brandboost_status == "success":   # LINE 202 — CRASH if PAGE_TARGET == "epic"
    print(f"✅ BrandBoost Studio Reel uploaded!")

Line 202 is outside the if PAGE_TARGET in ["brandboost", "both"] block.

If you run python3 fb-reel-upload.py video.mp4 --page epic, the script reaches

line 202 where brandboost_status has never been defined.

Python throws: UnboundLocalError: local variable 'brandboost_status' referenced before assignment

This is an indentation bug, but a good error-handling habit would have caught it

earlier: initialize variables before the conditional block:

brandboost_video_id, brandboost_status, brandboost_details = None, None, None

if PAGE_TARGET in ["brandboost", "both"]:
    brandboost_video_id, brandboost_status, brandboost_details = upload_video_to_fb(...)

if brandboost_status == "success":  # Now safe — worst case it's None
    print(f"✅ BrandBoost Studio Reel uploaded!")

---

The Exception Hierarchy

Python exceptions form a tree. When you write except Exception, you catch

almost everything. When you write except ValueError, you catch only value

errors. Specificity matters.

BaseException
├── KeyboardInterrupt      # Ctrl+C — almost never catch this
├── SystemExit             # sys.exit() — almost never catch this
└── Exception
    ├── ArithmeticError
    │   └── ZeroDivisionError
    ├── LookupError
    │   ├── KeyError       # dict["missing_key"]
    │   └── IndexError     # list[999]
    ├── TypeError          # wrong type passed to function
    ├── ValueError         # right type, wrong value (int("hello"))
    ├── OSError            # file system errors
    │   └── FileNotFoundError
    └── requests.exceptions.RequestException
        ├── ConnectionError
        ├── Timeout
        └── HTTPError

Catching Exception catches all of these. The problem: if you have a typo that

causes a NameError, you'll catch that too — hiding your bug.

The rule: catch the most specific exception you can handle.

---

Good vs Bad Exception Handling

Here's the pattern in fb-reel-upload.py (line 92):

# BAD — catches everything, converts to string, loses traceback
except Exception as e:
    print(f"⚠️  Upload error: {e}")
    return 'error', 'exception', str(e)

Now compare to trendpilot-loop.py's shopify_api_call (lines 143–161):

# GOOD — catches specific exceptions, handles each case differently
except urllib.error.HTTPError as e:
    if e.code == 429:
        retry_after = int(e.headers.get("Retry-After", "2"))
        log("API", f"  ⏳ Rate limited, retrying in {retry_after}s")
        time.sleep(retry_after)
        continue
    elif e.code == 401:
        return {"error": "Shopify token expired", "status": "token_expired"}
    else:
        err_body = e.read().decode()[:500] if e.fp else str(e)
        return {"error": f"Shopify API {e.code}: {err_body}", "status": "http_error"}
except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
    log("API", f"  ⚠️  Network error (attempt {attempt+1}): {e}")
    if attempt < max_retries - 1:
        time.sleep(2 ** attempt)  # Exponential backoff
        continue
    return {"error": f"Network error after {max_retries} retries: {e}"}

The trendpilot version:

• Catches `HTTPError` separately from network errors
• Handles 429 (rate limit) differently from 401 (bad token)
• Does exponential backoff (2^0=1s, 2^1=2s, 2^2=4s) before giving up
• Knows the difference between "retry" and "fail permanently"

---

Exponential Backoff

When an API is overloaded, retrying immediately makes it worse. Exponential

backoff waits longer after each failure:

for attempt in range(max_retries):
    try:
        result = call_api()
        return result
    except TemporaryError:
        if attempt < max_retries - 1:
            wait = 2 ** attempt   # 1s, 2s, 4s, 8s...
            time.sleep(wait)
        else:
            raise  # Re-raise after final attempt

trendpilot-loop.py does this correctly. daily-content-agent.py's qwen_call

has no retry at all — if Qwen is momentarily overloaded, you lose that article.

---

try / except / else / finally

You probably only use try/except. Python has two more clauses:

try:
    result = risky_operation()
except SomeError as e:
    print(f"Failed: {e}")
else:
    # Only runs if NO exception was raised
    print(f"Success: {result}")
    save_result(result)
finally:
    # ALWAYS runs, even if there's an exception
    cleanup()

The else clause is useful because it keeps "success" code separate from

"attempt" code. The finally clause is useful for cleanup (closing files,

releasing locks) that must happen regardless of outcome.

In amazon-price-tracker/main.py, add_product uses finally:

try:
    cursor.execute(...)
    conn.commit()
    return True
except sqlite3.Error as e:
    logger.error(f"Error adding product: {e}")
    return False
finally:
    conn.close()   # Always close the connection

---

The `raise` Statement

Three forms:

# 1. Raise a new exception
raise ValueError("price cannot be negative")

# 2. Re-raise the current exception (inside except block)
except Exception:
    log_it()
    raise   # Same exception continues up the stack

# 3. Chain exceptions (preserve original context)
except urllib.error.HTTPError as e:
    raise QwenError("API call failed") from e

The raise with no arguments (form 2) is underused. It lets you log or clean

up, then let the exception continue. You don't have to handle everything.

---

Exercises

Exercise 1 — Fix the crash bug:

Open scripts/fb-reel-upload.py. Add the two-line fix at line 202 that

initializes brandboost_video_id, brandboost_status, brandboost_details = None, None, None

before the if PAGE_TARGET in ["brandboost", "both"] block.

Then fix the indentation so the if brandboost_status == "success": check is

properly inside the block (or guard against None). Run it with --page epic to

verify it no longer crashes.

Exercise 2 — Add retry to qwen_call:

The qwen_call in daily-content-agent.py makes one attempt and gives up.

Add a retry loop (max 3 attempts) with exponential backoff for network errors.

Keep the no-retry behavior for DASHSCOPE_API_KEY not being set — that's not

a transient error.

Exercise 3 — Specific exceptions:

In daily-content-agent.py, load_extra_keywords() (line 457) has:

except Exception as e:
    print(f"  WARNING Could not load extra keywords: {e}")

What specific exceptions could happen here? (Hint: it opens a JSON file.)

Replace Exception with a tuple of the specific ones. Look up json.JSONDecodeError

and OSError.

Exercise 4 — finally for cleanup:

The token file reading in fb-reel-upload.py (line 16) opens a file without

closing it. Rewrite it using try/except/finally that closes the file even

if .strip() somehow fails. (Then note that with open(...) is even better —

that's the topic of Lesson 06.)

INTERMEDIATE — Lesson 03

Lesson 03 — Data Structures

Source files: scripts/daily-content-agent.py, scripts/trendpilot-loop.py

---

You Already Use the Right Ones

Before covering what to learn, acknowledge what you're already doing well:

• **Lists** for ordered sequences: keyword lists, trend lists, URL lists
• **Dicts** for key-value lookups: `KEYWORDS`, `product_info`, API payloads
• **Sets** for deduplication: `seen_ids`, `existing` in `load_extra_keywords`
• **Pathlib.Path** instead of string paths (better than most Python devs)

The goal here is knowing *when to choose what*, and learning two structures

you're not using yet: dataclasses and Counter/defaultdict.

---

The KEYWORDS Dict: A Good Pattern

daily-content-agent.py lines 43–188 define a large nested dict:

KEYWORDS = {
    "resumeforge": [
        "how to write a resume for warehouse workers",
        "resume summary examples for healthcare workers",
        # ... 30 more
    ],
    "clara": [...],
    "clozr": [...],
}

This is the right choice here. Why? Because you look things up by name

(KEYWORDS["resumeforge"]) and the order within each list matters (for rotation).

A list of tuples or a flat list would make the rotation logic harder.

The rotation logic at line 198–204 is elegant:

def pick_keywords(product, day_index, count=2):
    keywords = KEYWORDS.get(product, [])
    if not keywords:
        return []
    start = (day_index * count) % len(keywords)
    return [keywords[(start + i) % len(keywords)] for i in range(count)]

The % len(keywords) wraps around when you reach the end. Day 15 picks where

day 0 left off. This is a circular buffer — a useful pattern.

One improvement: KEYWORDS.get(product, []) is already defensive (returns []

if the product key doesn't exist). That's good.

---

The product_info Dict: Should Be a Dataclass

Inside write_article() (lines 360–397), you define a dict-of-dicts:

product_info = {
    "resumeforge": {
        "name": "ResumeForge",
        "url": "resumeforge.brandbooststudio.co",
        "angle": "AI resume builder...",
    },
    "clara": {
        "name": "Clara",
        "url": "clara.brandbooststudio.co",
        "phone": "(361) 734-4096",   # Only clara has this
        "angle": "AI phone receptionist...",
    },
    ...
}
info = product_info.get(product, {})
product_desc = info.get("angle", "")  # Empty string if missing — silently wrong

The problem: info.get("angle", "") silently returns empty string if you

typo the key. If you add a new field later, nothing tells you to add it to

all products. "clara" has a phone key that others don't — is that intentional?

A dataclass makes this self-documenting:

from dataclasses import dataclass
from typing import Optional

@dataclass
class ProductInfo:
    name: str
    url: str
    angle: str
    phone: Optional[str] = None  # Explicitly optional

PRODUCTS = {
    "resumeforge": ProductInfo(
        name="ResumeForge",
        url="resumeforge.brandbooststudio.co",
        angle="AI resume builder that refuses to fabricate experience.",
    ),
    "clara": ProductInfo(
        name="Clara",
        url="clara.brandbooststudio.co",
        angle="AI phone receptionist for small businesses.",
        phone="(361) 734-4096",
    ),
    # ...
}

# Usage
info = PRODUCTS.get(product)
if info is None:
    raise ValueError(f"Unknown product: {product}")
product_desc = info.angle   # Typo here → AttributeError immediately, not silent ""

Now if you typo info.naem, Python raises AttributeError immediately instead

of silently returning empty string.

---

Sets: Deduplication

You use sets correctly in load_extra_keywords() (line 467):

existing = set(KEYWORDS[product])
for kw in extra_kw:
    if kw not in existing:
        KEYWORDS[product].append(kw)
        existing.add(kw)

set() makes the kw not in existing check O(1) — instant, regardless of

list size. If you used if kw not in KEYWORDS[product] directly, that's O(n)

per check — fine for 30 keywords, slow for 30,000.

The pattern you're using (convert to set for lookups, write back to list to

preserve order) is exactly right.

---

Counter: Frequency Counts

You don't use Counter yet, but you will. If you ever want to know which

categories appear most in your trends:

from collections import Counter

categories = [t.get("category_en", "unknown") for t in trends]
counts = Counter(categories)

# Counter({'tech': 8, 'beauty': 5, 'outdoor': 3, 'pet': 2})
print(counts.most_common(3))  # Top 3 categories

In trendpilot-loop.py's _search_supplier_catalog, you manually score

each product with a loop. If you needed to count how often each category

appears in the catalog, Counter would do it in one line.

---

defaultdict: Skip the Key-Exists Check

You write this pattern in _record_post() (line 1427):

posted = {}
if POST_LOG.exists():
    try:
        with open(POST_LOG) as f:
            posted = json.load(f)
    except (json.JSONDecodeError, OSError):
        posted = {}

That's correct. But sometimes you need to accumulate into nested dicts and

you keep checking "does this key exist first?":

# Without defaultdict
category_counts = {}
for trend in trends:
    cat = trend.get("category_en", "unknown")
    if cat not in category_counts:
        category_counts[cat] = 0
    category_counts[cat] += 1

# With defaultdict
from collections import defaultdict
category_counts = defaultdict(int)  # Missing keys default to 0
for trend in trends:
    cat = trend.get("category_en", "unknown")
    category_counts[cat] += 1  # No key check needed

defaultdict(list) is also common when accumulating lists:

products_by_category = defaultdict(list)
for product in catalog:
    products_by_category[product["category"]].append(product)
# products_by_category["tech"] is automatically [] if you access it

---

When to Use What

Structure	Use when	Example in your code
-----------	----------	---------------------
`list`	Ordered, may have duplicates	keyword lists, trend list
`tuple`	Ordered, immutable, fixed size	return multiple values
`dict`	Key-value lookups, flexible keys	KEYWORDS, product_info
`set`	Membership testing, deduplication	`existing` in load_extra_keywords
`dataclass`	Structured records with named fields	Should replace product_info
`Counter`	Frequency counts	Category analysis
`defaultdict`	Dict where missing keys have a default	Grouping by category
`Enum`	Fixed set of named constants	ProductCategory in amazon-price-tracker

---

Exercises

Exercise 1 — Trace the rotation:

With len(KEYWORDS["resumeforge"]) == 30 and count=2:

• What keywords does day 0 return? (indices 0, 1)
• What keywords does day 15 return? (indices 30, 31 → wraps to 0, 1)
• What keywords does day 16 return?
• When does the rotation complete a full cycle?

Write out the math, then verify by calling pick_keywords("resumeforge", 0),

pick_keywords("resumeforge", 15), pick_keywords("resumeforge", 16).

Exercise 2 — Convert product_info to a dataclass:

Copy daily-content-agent.py's product_info dict (lines 360–397) and

convert it to a @dataclass called ProductInfo with fields: name, url,

angle, phone. Make phone optional with a default of None.

Create a module-level PRODUCTS dict that maps product keys to ProductInfo

instances. Update write_article() to use PRODUCTS instead of the local dict.

Exercise 3 — Use Counter:

In trendpilot-loop.py's analyze() function, after building the viable

list, add 3 lines that use Counter to print a summary: "Category breakdown:

tech: 4, beauty: 2, outdoor: 1". Import Counter from collections.

Exercise 4 — Set operations:

Python sets support union (|), intersection (&), and difference (-).

The word-overlap check in trendpilot-loop.py line 426–430:

name_words = set(w.lower() for w in name.split() if len(w) > 2 and w.lower() not in skip_words)
title_words = set(w.lower() for w in title_lower.split() if len(w) > 2 and w.lower() not in skip_words)
overlap = len(name_words & title_words) / max(len(name_words), 1)

name_words & title_words is a set intersection. len(...) gives the count of

shared words. Can you extend this to also compute the words that are in the

title but NOT in the name? That's title_words - name_words. Try it in a Python

interpreter.

INTERMEDIATE — Lesson 04

Lesson 04 — Classes & OOP

Source files: All four scripts

---

Three Levels of Organization in Your Code

Your four scripts show three different approaches to organizing code:

Level 1 — Pure functions (daily-content-agent.py)

Everything is a standalone function. write_article() takes arguments, does

work, returns a value. No shared state except module-level globals. This is

actually fine for scripts that do one job.

Level 2 — Thin class wrapper (trendpilot-loop.py)

TrendPilotLoop is a class, but all the interesting code is in module-level

functions (observe, analyze, source, etc.). The class just holds dry_run

and results and delegates. This is a transitional pattern.

Level 3 — Full OOP (amazon-price-tracker/main.py)

AmazonPriceMonitor holds all state (db_path, api_key, config, running)

and all behavior (add_product, fetch_amazon_price, run_monitor). The class

is the unit of organization.

Neither level is objectively better. The question is: *what problem does the

class solve?*

---

When a Class Helps

A class helps when you have:

1. State that persists between function calls — the monitor needs to remember

its running flag, its config, its db_path

2. Multiple related functions that share the same data — every method in

AmazonPriceMonitor needs self.db_path, self.api_key, self.config

3. Multiple instances — you could create two monitors watching different

databases: us_monitor = AmazonPriceMonitor("us.db") and

eu_monitor = AmazonPriceMonitor("eu.db")

Compare to daily-content-agent.py: there's no state to share. KEYWORDS is

a constant. Each function call is independent. A class would add complexity

without benefit.

---

Anatomy of a Class

class AmazonPriceMonitor:
    # __init__ is called when you do AmazonPriceMonitor()
    # "self" refers to this specific instance
    def __init__(self, config_path: str = "config.json"):
        self.db_path = "amazon_price_monitor.db"   # instance attribute
        self.api_key = os.getenv('RAPID_API_KEY')   # instance attribute
        self.running = True                          # instance attribute
        self.setup_database()                        # call another method
    
    # A method — just a function that gets "self" as first argument
    def add_product(self, product: Dict) -> bool:
        conn = sqlite3.connect(self.db_path)  # uses self.db_path
        # ...
    
    def run_monitor(self):
        while self.running:  # checks self.running
            # ...
    
    def stop_monitor(self):
        self.running = False  # modifies self.running — run_monitor sees this

self is just the name for "this instance." It's passed automatically —

you don't pass it when calling methods:

monitor = AmazonPriceMonitor()   # __init__ called with self=monitor
monitor.add_product(data)         # add_product called with self=monitor
monitor.stop_monitor()            # stop_monitor called with self=monitor

---

What TrendPilotLoop Does Wrong

TrendPilotLoop in trendpilot-loop.py (line 1629):

class TrendPilotLoop:
    def __init__(self, dry_run=False):
        self.dry_run = dry_run
        self.log_entries = []
        self.results = {}

    def run_full_cycle(self, ...):
        # calls module-level functions: observe(), analyze(), source()...
        # stores results in self.results

The problem: self.results is set at the end of run_full_cycle but self.log_entries

is never actually used anywhere in the class. The "results" are just stored in a

dict that callers then have to reach into. The module-level functions like

observe() and analyze() have no access to self.dry_run — they receive it

as a parameter.

If you were going to use a class here, either:

• Move `observe`, `analyze`, `source`, etc. *into* the class as methods
• Or skip the class entirely and just use functions

The hybrid approach causes confusion.

---

Writing a Useful Class: ContentAgent

Here's how you might refactor daily-content-agent.py into a class *if* you

wanted to manage state (like tracking which articles have been written):

class ContentAgent:
    def __init__(self, workspace: Path, api_key: str):
        self.workspace = workspace
        self.api_key = api_key
        self.blog_dir = workspace / "blog-drafts"
        self.articles_written = 0   # Track progress this run
        self._day_index = None      # Lazy-compute once
    
    @property
    def day_index(self) -> int:
        """Compute day index once and cache it."""
        if self._day_index is None:
            start_date = datetime(2026, 7, 2)
            self._day_index = max(0, (datetime.now() - start_date).days)
        return self._day_index
    
    def pick_keywords(self, product: str, count: int = 2) -> list[str]:
        keywords = KEYWORDS.get(product, [])
        if not keywords:
            return []
        start = (self.day_index * count) % len(keywords)
        return [keywords[(start + i) % len(keywords)] for i in range(count)]
    
    def write_article(self, product: str, keyword: str) -> Path:
        # ... can use self.api_key, self.blog_dir, self.articles_written
        self.articles_written += 1
        return file_path
    
    def run(self, products: list[str]):
        for product in products:
            keywords = self.pick_keywords(product)
            for keyword in keywords:
                self.write_article(product, keyword)
        print(f"Written {self.articles_written} articles")

The @property decorator (line 3) is covered in Lesson 08, but notice it lets

you call agent.day_index like an attribute (no ()) while still computing it

the first time. The result is cached in self._day_index.

---

Inheritance (Brief)

Classes can inherit from other classes. amazon-price-tracker doesn't use it,

but the exception we created in Lesson 01 does:

class QwenError(Exception):
    pass

QwenError inherits from Exception. It gets all of Exception's behavior

(can be raised, caught, has a message) but is its own distinct type. You can

except QwenError without catching all exceptions.

You don't need deep inheritance hierarchies for your use cases. A one-level

"inherit from Exception to make a custom error type" is the most useful form.

---

Class vs Module

Python modules (.py files) are also a unit of organization. A file full of

related functions is valid design. Don't make everything a class just because

you learned OOP.

Rule of thumb:

• One-off script that runs start to finish → functions in a module is fine
• Tool that runs repeatedly, keeps state, or might have multiple instances → class
• Agent with phases that share state (TrendPilot) → class makes sense
• Content writer with independent calls → functions are fine

---

Exercises

Exercise 1 — Read and understand:

Open amazon-price-tracker/main.py. Find where threading and Queue are

imported (line 17–18) but never used in the class. This is dead code.

Look at run_monitor() (line 525). What would you need to change to use a

Queue to receive "stop" signals from another thread instead of the self.running

flag? (Sketch it, don't implement it.)

Exercise 2 — Add a class attribute:

In AmazonPriceMonitor, check_interval = 300 is set in __init__. Move it

to a *class attribute* (outside __init__, at the class level):

class AmazonPriceMonitor:
    CHECK_INTERVAL = 300   # class attribute — shared by all instances
    
    def __init__(self, config_path: str = "config.json"):
        self.check_interval = self.config.get("check_interval", self.CHECK_INTERVAL)

What's the difference between self.check_interval (instance attribute) and

AmazonPriceMonitor.CHECK_INTERVAL (class attribute)?

Exercise 3 — Create a class:

trendpilot-loop.py's _already_posted and _record_post functions both read

and write the same JSON file. Create a PostLog class that:

• Takes a `log_file: Path` in `__init__`
• Loads the JSON into `self.data` in `__init__`
• Has an `already_posted(handle, name) -> bool` method
• Has a `record(handle, name, post_id)` method that writes back to disk
• Has a `cleanup(days=30)` method that removes old entries

Replace the two module-level functions with this class. Instantiate it once at

the top of publish().

Exercise 4 — Think about it:

daily-content-agent.py runs as a cron job once per day. Do you actually need

a class for it? What would be the benefit? What would be the cost? Write 2–3

sentences with your opinion.

INTERMEDIATE — Lesson 05

Lesson 05 — CLI & argparse

Source files: scripts/fb-reel-upload.py vs scripts/trendpilot-loop.py

---

Two Ways to Parse Arguments

fb-reel-upload.py parses command-line arguments manually (lines 21–37):

VIDEO_PATH = sys.argv[1] if len(sys.argv) > 1 else None
MODE = sys.argv[2] if len(sys.argv) > 2 else "slideshow"
MODE = MODE.lower()

if "--caption" in sys.argv:
    idx = sys.argv.index("--caption")
    if idx + 1 < len(sys.argv):
        CAPTION_OVERRIDE = sys.argv[idx + 1]

if "--page" in sys.argv:
    idx = sys.argv.index("--page")
    if idx + 1 < len(sys.argv):
        PAGE_TARGET = sys.argv[idx + 1].lower()

trendpilot-loop.py uses argparse (lines 1835–1844):

parser = argparse.ArgumentParser(description="TrendPilot AI — Closed-Loop Autopilot")
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--observe-only", action="store_true")
parser.add_argument("--limit", type=int, default=20)
args = parser.parse_args()

The manual approach breaks silently in ways that argparse prevents.

---

The Problem with Manual sys.argv Parsing

Problem 1: Positional argument collision

VIDEO_PATH = sys.argv[1] if len(sys.argv) > 1 else None
MODE = sys.argv[2] if len(sys.argv) > 2 else "slideshow"

This assumes arg 1 is always the video path and arg 2 is always the mode.

But what if someone writes:

python3 fb-reel-upload.py --page epic video.mp4

Now sys.argv[1] is "--page" (not the video path) and sys.argv[2] is

"epic" (not the mode). The script silently tries to upload a file named

--page.

Problem 2: No --help

Run python3 fb-reel-upload.py --help. You get the Python default help

message (useless). With argparse, --help is automatically generated from

your argument definitions.

Problem 3: No type validation

--limit in trendpilot uses type=int. If you pass --limit foo, argparse

prints: error: argument --limit: invalid int value: 'foo'. With manual

parsing, you'd get a ValueError deep inside the script.

Problem 4: No required-argument enforcement

If you forget to pass VIDEO_PATH, the script crashes at line 161 with

"❌ Video not found: None". That's not a helpful error. argparse would have

said: error: the following arguments are required: video_path.

---

argparse Rewrite of fb-reel-upload.py

import argparse

def parse_args():
    parser = argparse.ArgumentParser(
        description="Upload video as Facebook Reel",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s video.mp4
  %(prog)s video.mp4 --slideshow --page epic
  %(prog)s video.mp4 --caption "Custom caption here"
        """
    )
    
    parser.add_argument(
        "video_path",
        help="Path to the video file to upload"
    )
    parser.add_argument(
        "--mode",
        choices=["slideshow", "scene"],
        default="slideshow",
        help="Caption style (default: slideshow)"
    )
    parser.add_argument(
        "--caption",
        help="Override the default caption"
    )
    parser.add_argument(
        "--page",
        choices=["epic", "brandboost", "both"],
        default="both",
        help="Which Facebook page to upload to (default: both)"
    )
    
    return parser.parse_args()


def main():
    args = parse_args()
    
    # Now use args.video_path, args.mode, args.caption, args.page
    # No positional collision possible
    if not os.path.exists(args.video_path):
        print(f"❌ Video not found: {args.video_path}")
        sys.exit(1)
    
    caption = args.caption or CAPTIONS[args.mode]
    # ...

---

argparse Reference

Positional arguments (required, no --):

parser.add_argument("video_path", help="Path to video file")
# args.video_path

Optional flags (optional, have --):

parser.add_argument("--dry-run", action="store_true", help="Preview only")
# args.dry_run (note: hyphen becomes underscore)

Optional with value (optional, but requires a value if present):

parser.add_argument("--limit", type=int, default=20, help="Max items")
# args.limit → 20 by default, or int you passed

Choices (must be one of a fixed set):

parser.add_argument("--page", choices=["epic", "brandboost", "both"], default="both")

Short forms (common):

parser.add_argument("-n", "--dry-run", action="store_true")
# Either -n or --dry-run works

---

Separating Logic from the CLI

A good pattern: put your actual logic in functions, and keep main() thin:

# The logic — testable, importable
def upload_video(video_path: str, mode: str, caption: str, page: str):
    ...

# The CLI — parses args, calls logic
def main():
    args = parse_args()
    caption = args.caption or CAPTIONS[args.mode]
    upload_video(args.video_path, args.mode, caption, args.page)

if __name__ == "__main__":
    main()

This matters because:

1. You can test upload_video() without going through the CLI

2. Another script can from fb_reel_upload import upload_video and call it directly

3. main() stays readable — it's just "parse, then call"

---

sys.exit() and Return Codes

Your scripts use sys.exit(1) correctly when they fail. The convention:

• `0` means success
• Non-zero means failure (the specific number can encode the error type)
• `sys.exit(1)` is the standard "something went wrong" exit

Shell scripts and cron jobs check this:

python3 fb-reel-upload.py video.mp4
if [ $? -ne 0 ]; then
    echo "Upload failed!"
fi

---

Exercises

Exercise 1 — Add --help to fb-reel-upload.py:

Rewrite the argument parsing section (lines 20–38) using argparse. Keep

all the same flags (--caption, --page) but now add them properly.

Run python3 fb-reel-upload.py --help and verify the help text is generated.

Exercise 2 — Fix the positional arg bug:

With your argparse version, test that this no longer breaks:

python3 fb-reel-upload.py --page epic cleaneats-tiktok-final.mp4

With the old sys.argv approach, this tries to upload a file called --page.

With argparse, it correctly identifies cleaneats-tiktok-final.mp4 as the video.

Exercise 3 — Add a subcommand:

trendpilot-loop.py has --observe-only, --learn-only, --source-only, etc.

These are flags that really should be *subcommands*. Rewrite the CLI to use

subcommands:

python3 trendpilot-loop.py observe
python3 trendpilot-loop.py learn
python3 trendpilot-loop.py run --dry-run --limit 5

Research parser.add_subparsers().

Exercise 4 — Type coercion:

In trendpilot-loop.py, --limit uses type=int. What happens if you run:

python3 trendpilot-loop.py --limit 3.5

What if you run:

python3 trendpilot-loop.py --limit five

Try both. What does argparse print? How would you add a validator to reject

limits below 1 or above 100?

INTERMEDIATE — Lesson 06

Lesson 06 — File I/O & Context Managers

Source files: scripts/fb-reel-upload.py, scripts/daily-content-agent.py

---

The Unclosed File Bug

Line 16 of fb-reel-upload.py:

FB_EPICS_TOKEN = open('/home/ron/.openclaw/workspace/.fb_page_token').read().strip()

Two problems:

1. The file handle is never closed — it leaks until Python's garbage collector

gets around to it (nondeterministic)

2. If .fb_page_token doesn't exist, you get FileNotFoundError at *import

time*, before any helpful error message prints

The fix for both is the with statement (context manager):

try:
    with open('/home/ron/.openclaw/workspace/.fb_page_token') as f:
        FB_EPICS_TOKEN = f.read().strip()
except FileNotFoundError:
    print("ERROR: .fb_page_token not found. Run setup first.")
    sys.exit(1)

The with statement guarantees f.close() is called even if an exception

occurs mid-read. It's the Python idiom for any resource that needs cleanup.

---

How Context Managers Work

# This:
with open("file.txt") as f:
    data = f.read()
# f is closed here, guaranteed

# Is equivalent to:
f = open("file.txt")
try:
    data = f.read()
finally:
    f.close()

The with version is shorter and impossible to get wrong. Always use with

when opening files.

You can open multiple files in one with:

with open("input.txt") as src, open("output.txt", "w") as dst:
    dst.write(src.read())

---

File Modes

open("file.txt")        # "r" default — read text, error if not found
open("file.txt", "r")   # same
open("file.txt", "w")   # write text, creates if not found, TRUNCATES if exists
open("file.txt", "a")   # append text, creates if not found, does NOT truncate
open("file.txt", "rb")  # read binary (for images, PDFs, etc.)
open("file.txt", "wb")  # write binary
open("file.txt", "x")   # create new, error if already exists (safe write)

Your fb-reel-upload.py log at line 213:

with open('/home/ron/.openclaw/workspace/logs/fb-reel-upload.log', 'a') as f:
    f.write(f"\n[{datetime.now().isoformat()}]\n")

The 'a' mode is correct here — appending to the log without erasing it.

---

pathlib.Path vs Strings

You use pathlib.Path well in daily-content-agent.py:

WORKSPACE = Path(__file__).parent.parent
BLOG_DIR = WORKSPACE / "blog-drafts"
LOG_FILE = WORKSPACE / "logs" / "daily-content.log"

But fb-reel-upload.py uses raw strings:

open('/home/ron/.openclaw/workspace/.fb_page_token')
open('/home/ron/.openclaw/workspace/logs/fb-reel-upload.log', 'a')

The hardcoded path means the script breaks if you move the workspace. Better:

from pathlib import Path

WORKSPACE = Path(__file__).parent.parent
TOKEN_FILE = WORKSPACE / ".fb_page_token"
LOG_FILE = WORKSPACE / "logs" / "fb-reel-upload.log"

try:
    FB_EPICS_TOKEN = TOKEN_FILE.read_text().strip()
except FileNotFoundError:
    print(f"ERROR: Token file not found: {TOKEN_FILE}")
    sys.exit(1)

Path.read_text() opens, reads, and closes in one call. Equivalent to:

with open(TOKEN_FILE) as f:
    FB_EPICS_TOKEN = f.read().strip()

---

Useful pathlib Methods

You're already using many of these, but for completeness:

p = Path("/home/ron/.openclaw/workspace/scripts/daily-content-agent.py")

p.exists()          # True/False
p.is_file()         # True
p.is_dir()          # False
p.name              # "daily-content-agent.py"
p.stem              # "daily-content-agent"
p.suffix            # ".py"
p.parent            # Path("/home/ron/.openclaw/workspace/scripts")
p.parent.parent     # Path("/home/ron/.openclaw/workspace")

# Reading/writing
p.read_text()                # file contents as string
p.read_bytes()               # file contents as bytes
p.write_text("content")      # write string (truncates first)
p.write_bytes(b"content")    # write bytes

# Navigation
WORKSPACE = Path("/home/ron/.openclaw/workspace")
WORKSPACE / "scripts" / "my.py"  # Path("/home/ron/.openclaw/workspace/scripts/my.py")

# Listing
list(WORKSPACE.glob("*.py"))               # all .py files in workspace
list(WORKSPACE.glob("**/*.py"))            # all .py files recursively
list(WORKSPACE.rglob("*.py"))              # same
sorted(DATA_DIR.glob("*.json"), reverse=True)  # you do this in trendpilot

# Making directories
p.mkdir(parents=True, exist_ok=True)   # you do this correctly in trendpilot

# Stat
p.stat().st_mtime    # modification time (seconds since epoch)

---

Reading JSON Files Safely

Pattern from trendpilot-loop.py's _record_post (lines 1432–1436):

posted = {}
if POST_LOG.exists():
    try:
        with open(POST_LOG) as f:
            posted = json.load(f)
    except (json.JSONDecodeError, OSError):
        posted = {}

This is the right pattern. The two exceptions you need:

• `json.JSONDecodeError` — file exists but isn't valid JSON (truncated, empty, corrupted)
• `OSError` — file exists but can't be read (permissions, etc.)

You can also use Path.read_text():

try:
    posted = json.loads(POST_LOG.read_text())
except (json.JSONDecodeError, OSError, FileNotFoundError):
    posted = {}

FileNotFoundError is a subclass of OSError, so catching OSError covers it.

But being explicit about FileNotFoundError makes the code more readable.

---

Writing JSON Safely

The naive write:

with open(path, "w") as f:
    json.dump(data, f)

Problem: if the process crashes mid-write, you get a partial/corrupted file.

If this file was your posted_products.json, you'd lose your deduplication state.

Safer pattern using a temp file + atomic rename:

import tempfile

def write_json_safe(path: Path, data: dict):
    """Write JSON to a temp file, then rename — atomic on Linux."""
    tmp = path.with_suffix('.tmp')
    try:
        with open(tmp, "w") as f:
            json.dump(data, f, indent=2)
        tmp.rename(path)   # atomic on same filesystem
    except Exception:
        tmp.unlink(missing_ok=True)  # cleanup on failure
        raise

This matters for files that track important state (like your post log or

keyword expansion marker). For blog articles, the naive write is fine.

---

The .env File Parsing You're Doing Manually

daily-content-agent.py lines 32–40:

if not DASHSCOPE_API_KEY:
    env_file = WORKSPACE / ".env"
    if env_file.exists():
        with open(env_file) as f:
            for line in f:
                if line.strip().startswith("DASHSCOPE_API_KEY="):
                    DASHSCOPE_API_KEY = line.strip().split("=", 1)[1]
                    break

This works, but doesn't handle:

• Quoted values: `DASHSCOPE_API_KEY="abc123"` (your code would include the quotes)
• Comments after values: `DASHSCOPE_API_KEY=abc123 # my key`
• Multiline values

The python-dotenv library handles all of this:

pip install python-dotenv

from dotenv import load_dotenv

load_dotenv(WORKSPACE / ".env")   # loads all variables into os.environ
DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY", "")

But: your current code works for your use case. Don't add a dependency unless

it solves an actual problem you have.

---

Exercises

Exercise 1 — Fix fb-reel-upload.py token loading:

Lines 16–19 open token files without with and without error handling.

Rewrite them using pathlib and proper error handling. If a token file doesn't

exist, print a helpful message and exit cleanly instead of crashing.

Exercise 2 — Fix fb-reel-upload.py hardcoded paths:

Define WORKSPACE = Path(__file__).parent.parent at the top of the file.

Replace all hardcoded /home/ron/.openclaw/workspace/... paths with

WORKSPACE / "..." equivalents.

Exercise 3 — Trace the glob:

In trendpilot-loop.py line 213:

live_files = sorted(DATA_DIR.glob("dropship_trends_*.json"), reverse=True)

What does * match in a glob pattern? What would **/*.json match? Open a

Python shell and try:

from pathlib import Path
list(Path(".").glob("*.py"))

Exercise 4 — Atomic write:

Your _record_post function in trendpilot writes the JSON directly:

with open(POST_LOG, "w") as f:
    json.dump(posted, f, indent=2)

Rewrite it using the write_json_safe pattern from this lesson. Test that

if you interrupt it mid-write, the original file is not corrupted.

(Hint: you can simulate a crash with raise RuntimeError("crash") inside the

write, before the rename.)

INTERMEDIATE — Lesson 07

Lesson 07 — Working with APIs

Source files: scripts/daily-content-agent.py, scripts/trendpilot-loop.py

---

The Problem: Two Scripts, Two Completely Different API Styles

You make API calls in both scripts. Compare them:

daily-content-agent.py lines 212–238 — bare-bones, no retry:

def qwen_call(prompt, model="qwen-plus", temperature=0.7, max_tokens=4000):
    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = { ... }
    try:
        req = urllib.request.Request(QWEN_API_URL, data=json.dumps(payload).encode(),
                                      headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=120) as resp:
            data = json.loads(resp.read())
            return data["choices"][0]["message"]["content"]
    except Exception as e:
        return {"error": str(e)}   # ← returns a dict on error (see Lesson 01)

trendpilot-loop.py lines 114–163 — retry, rate limit handling, specific errors:

def shopify_api_call(url, method="GET", data=None, headers=None, max_retries=3):
    for attempt in range(max_retries):
        try:
            req = urllib.request.Request(url, ...)
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as e:
            if e.code == 429:
                retry_after = int(e.headers.get("Retry-After", "2"))
                time.sleep(retry_after)
                continue
            elif e.code == 401:
                return {"error": "Shopify token expired", "status": "token_expired"}
            else:
                return {"error": f"Shopify API {e.code}: ...", "status": "http_error"}
        except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)   # exponential backoff
                continue
            return {"error": f"Network error after {max_retries} retries: {e}"}

shopify_api_call is much better. But it still returns error dicts instead of raising.

And neither version handles one important case: what does the JSON actually contain?

---

How HTTP APIs Actually Work

Every API call is:

1. You send an HTTP request (method + URL + headers + body)

2. The server sends back a response (status code + body)

3. You parse the body (usually JSON)

4. You handle errors (bad status codes, network failures, unexpected JSON shape)

Your script ──POST──▶ https://dashscope-intl.aliyuncs.com/...
                              │
              ◀──200 OK──────┘
              body: {"choices": [{"message": {"content": "article text"}}]}

Status codes that matter:

• `200` — success
• `400` — you sent bad data (check your payload)
• `401` — bad or expired API key
• `403` — forbidden (wrong permissions)
• `429` — rate limited (slow down)
• `500` — server error on their side (retry)
• `503` — service temporarily unavailable (retry)

---

urllib.request vs requests

Your scripts use urllib.request (built into Python — no install required).

Most tutorials use the requests library (requires pip install requests).

# urllib.request (what your scripts use)
import urllib.request, json
req = urllib.request.Request(url, data=json.dumps(payload).encode(),
                              headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=30) as resp:
    data = json.loads(resp.read())

# requests (simpler syntax)
import requests
resp = requests.post(url, json=payload, headers=headers, timeout=30)
resp.raise_for_status()   # raises HTTPError for 4xx/5xx
data = resp.json()

Key difference: requests.raise_for_status() auto-raises on bad status codes.

With urllib, a 429 or 500 raises urllib.error.HTTPError — which you're already catching.

The requests version is cleaner for new code. For your existing scripts, stick with

urllib unless you're already adding dependencies — it works fine.

---

The Tricky Part: Parsing the Response

This line in your Qwen call is doing a lot:

return data["choices"][0]["message"]["content"]

What can go wrong:

• `data` might be `{}` if the API returned an unexpected shape
• `data["choices"]` might be `[]` (empty list)
• `data["choices"][0]["message"]` might not have `"content"`

The trendpilot-loop.py version handles this better (line 193):

content = result.get("choices", [{}])[0].get("message", {}).get("content", "")

This uses .get() with defaults so it never crashes on a bad response — it returns

an empty string instead. But an empty string might silently write a blank article.

Better: detect the empty case and raise:

def _parse_qwen_response(data: dict) -> str:
    try:
        content = data["choices"][0]["message"]["content"]
        if not content:
            raise ValueError(f"Empty content in response: {data}")
        return content
    except (KeyError, IndexError) as e:
        raise ValueError(f"Unexpected Qwen response shape: {data}") from e

---

A Production-Grade API Call Function

Here's what qwen_call looks like when you apply everything from this lesson

(plus Lesson 01's exception principle and Lesson 07's retry pattern):

import urllib.request, urllib.error
import json, time

class QwenError(Exception):
    pass

def qwen_call(prompt, model="qwen-plus", temperature=0.7, max_tokens=4000,
              max_retries=3) -> str:
    """Call Qwen API. Always returns str. Raises QwenError on failure."""
    if not DASHSCOPE_API_KEY:
        raise QwenError("DASHSCOPE_API_KEY not set")

    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "max_tokens": max_tokens,
        "stream": False,
    }

    last_err = None
    for attempt in range(max_retries):
        try:
            req = urllib.request.Request(
                QWEN_API_URL,
                data=json.dumps(payload).encode(),
                headers=headers,
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=120) as resp:
                data = json.loads(resp.read())
                content = data["choices"][0]["message"]["content"]
                if not content:
                    raise QwenError("Empty response from Qwen")
                return content

        except urllib.error.HTTPError as e:
            body = e.read().decode()[:300]
            if e.code == 429:
                wait = int(e.headers.get("Retry-After", 2 ** attempt))
                print(f"  Qwen rate limited, waiting {wait}s (attempt {attempt+1})")
                time.sleep(wait)
                last_err = QwenError(f"Rate limited: {body}")
                continue
            elif e.code in (500, 503):
                time.sleep(2 ** attempt)
                last_err = QwenError(f"HTTP {e.code}: {body}")
                continue
            else:
                raise QwenError(f"HTTP {e.code}: {body}") from e

        except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
            wait = 2 ** attempt
            print(f"  Qwen network error, retrying in {wait}s: {e}")
            time.sleep(wait)
            last_err = QwenError(str(e))

    raise last_err or QwenError("Max retries exceeded")

What changed from your current version:

• Raises `QwenError` instead of returning error dicts
• Retries on 429 (rate limit) and 5xx (server errors) using exponential backoff
• Retries on network/timeout errors
• Validates that `content` is non-empty before returning
• Has a `-> str` return type annotation

---

Reading API Keys Safely

Your scripts read API keys from .env files with this pattern:

DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY", "")
if not DASHSCOPE_API_KEY:
    env_file = WORKSPACE / ".env"
    if env_file.exists():
        with open(env_file) as f:
            for line in f:
                if line.strip().startswith("DASHSCOPE_API_KEY="):
                    DASHSCOPE_API_KEY = line.strip().split("=", 1)[1]
                    break

This works. The only issue: if .env has DASHSCOPE_API_KEY="value" with quotes,

you'd include the quotes in the key. Add a .strip('"\''):

DASHSCOPE_API_KEY = line.strip().split("=", 1)[1].strip().strip('"\'')

Or use the python-dotenv package: pip install python-dotenv and

from dotenv import load_dotenv; load_dotenv() — it handles all the edge cases.

---

Exercises

Exercise 1 — Read the retry logic:

In trendpilot-loop.py, find shopify_api_call (around line 114). Draw a flowchart

of what happens on:

• A 429 error on the first attempt
• A 401 error on any attempt
• Three consecutive network errors

Exercise 2 — Test the response parser:

Write a function parse_qwen_response(data: dict) -> str that handles:

• Normal response: `{"choices": [{"message": {"content": "hello"}}]}`
• Empty content: `{"choices": [{"message": {"content": ""}}]}`
• Missing keys: `{"error": "quota exceeded"}`

What should each case return or raise?

Exercise 3 — Add retry to qwen_call:

In daily-content-agent.py, modify qwen_call to retry up to 3 times on

urllib.error.HTTPError with code 429, with a 5-second wait between attempts.

Don't change anything else — just add the retry loop.

Exercise 4 — Identify what status codes your APIs return:

Pick one of your API integrations (Shopify, Qwen, Facebook). Look at the code

and list which HTTP status codes you currently handle explicitly vs. which ones

fall through to the generic except Exception handler. What would happen if

Qwen returned a 503?

INTERMEDIATE — Lesson 08

Lesson 08 — Decorators and Context Managers

Source files: scripts/trendpilot-loop.py, scripts/daily-content-agent.py

---

The Problem: Copy-Pasted Retry Logic

trendpilot-loop.py has retry logic in shopify_api_call (lines 138–163).

daily-content-agent.py's qwen_call has no retry at all.

If you wanted retry in three different functions, you'd copy the loop each time.

That's three places to fix if you change the backoff formula. Decorators solve this.

---

What Is a Decorator?

A decorator is a function that wraps another function to add behavior before

or after it runs — without changing the function itself.

You already use them without thinking about it. In Lesson 04's OOP example:

@property
def is_profitable(self):
    return self.margin > 0.3

@property is a built-in decorator. It wraps is_profitable so accessing

product.is_profitable runs the function instead of returning the function object.

The @ syntax is just shorthand:

# These two are identical:

@some_decorator
def my_function():
    ...

# Same as:
def my_function():
    ...
my_function = some_decorator(my_function)

---

Writing a Retry Decorator

Here's the retry logic that's currently inside shopify_api_call, extracted into

a decorator:

import time
import functools

def retry(max_attempts=3, wait=2, exceptions=(Exception,)):
    """Decorator: retry a function up to max_attempts times on exception."""
    def decorator(func):
        @functools.wraps(func)       # preserves __name__ and __doc__
        def wrapper(*args, **kwargs):
            last_err = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_err = e
                    if attempt < max_attempts - 1:
                        sleep_time = wait * (2 ** attempt)  # exponential backoff
                        print(f"  Retry {attempt+1}/{max_attempts} for {func.__name__}: {e}")
                        time.sleep(sleep_time)
            raise last_err
        return wrapper
    return decorator

Now instead of manually writing retry loops, you decorate:

import urllib.error

@retry(max_attempts=3, wait=1, exceptions=(urllib.error.URLError, TimeoutError))
def qwen_call(prompt, model="qwen-plus", temperature=0.7, max_tokens=4000) -> str:
    headers = {"Authorization": f"Bearer {DASHSCOPE_API_KEY}", ...}
    req = urllib.request.Request(QWEN_API_URL, ...)
    with urllib.request.urlopen(req, timeout=120) as resp:
        data = json.loads(resp.read())
        return data["choices"][0]["message"]["content"]

The function body contains only the happy path. The retry logic lives in one place.

---

How the Decorator Stack Works

@functools.wraps(func) is important. Without it:

print(qwen_call.__name__)   # → "wrapper"  ← wrong, hard to debug
print(qwen_call.__doc__)    # → None        ← lost the docstring

# With @functools.wraps:
print(qwen_call.__name__)   # → "qwen_call" ← correct
print(qwen_call.__doc__)    # → "Call Qwen Cloud API..." ← preserved

Always use @functools.wraps(func) when writing decorators.

---

A Timing Decorator

Another common one — seeing how long your API calls take:

import time
import functools

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"  {func.__name__} took {elapsed:.2f}s")
        return result
    return wrapper

@timed
def qwen_call(prompt, ...):
    ...

# Output: "  qwen_call took 4.37s"

You can stack decorators:

@timed
@retry(max_attempts=3, wait=1, exceptions=(urllib.error.URLError,))
def qwen_call(prompt, ...):
    ...

They apply bottom-up: qwen_call is first wrapped by retry, then the result is

wrapped by timed. So timed sees the total time including retries.

---

Context Managers: Beyond Files

Lesson 06 covered with open(...) for files. Context managers are the same

pattern for anything that has setup and teardown:

# File — you already use this
with open("data.json") as f:
    data = json.load(f)

# Database connection
with sqlite3.connect("db.sqlite") as conn:
    conn.execute("INSERT INTO ...")
# conn.close() called automatically

# Timing (stdlib)
import contextlib, time

@contextlib.contextmanager
def timer(label):
    start = time.perf_counter()
    yield                          # code inside the with block runs here
    elapsed = time.perf_counter() - start
    print(f"{label}: {elapsed:.2f}s")

with timer("Qwen call"):
    article = qwen_call(prompt)
# prints: "Qwen call: 4.37s"

---

Writing Your Own Context Manager

@contextlib.contextmanager lets you write a context manager as a generator:

import contextlib
from pathlib import Path

@contextlib.contextmanager
def atomic_write(path: Path):
    """Write to a temp file, then rename to final path on success.
    
    If the write fails halfway, the original file is untouched.
    """
    tmp_path = path.with_suffix(path.suffix + ".tmp")
    try:
        with open(tmp_path, "w") as f:
            yield f           # caller writes to f here
        tmp_path.rename(path) # rename is atomic on Linux
    except:
        tmp_path.unlink(missing_ok=True)
        raise

# Usage:
with atomic_write(file_path) as f:
    f.write(article)
# If qwen_call failed before we got here, the original file is intact

This is useful in your write_article() function — if the HTML generation crashes

after you've already written the markdown, you'd want atomic behavior.

---

Applying Both to Your Scripts

Here's the pattern in daily-content-agent.py for writing each article:

# Current code (daily-content-agent.py ~line 570)
for keyword in keywords:
    file_path = write_article(product, keyword, day_index, article_num)

With decorators and context managers, write_article becomes:

@timed
@retry(max_attempts=2, wait=5, exceptions=(QwenError,))
def write_article(product: str, keyword: str, day_index: int, article_num: int) -> str:
    slug = slugify(keyword)
    file_path = BLOG_DIR / f"{product}-{slug}.md"

    prompt = build_prompt(product, keyword)   # extracted into its own function
    article = qwen_call(prompt)               # raises QwenError on failure

    with atomic_write(file_path) as f:        # atomic — no half-written files
        f.write(article)

    return str(file_path)

Each concern is handled in one place:

• Timing: `@timed` decorator
• Retry: `@retry` decorator
• File safety: `atomic_write` context manager
• The function body: just the happy path

---

Exercises

Exercise 1 — Understand the mechanics:

Read this and predict the output before running it:

def shout(func):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"Done with {func.__name__}")
        return result
    return wrapper

@shout
def add(a, b):
    return a + b

print(add(2, 3))

Then run it. Were you right?

Exercise 2 — Write the retry decorator:

Copy the retry decorator from this lesson into a file scripts/utils.py.

Add a test at the bottom:

attempt_count = 0

@retry(max_attempts=3, wait=0, exceptions=(ValueError,))
def flaky():
    global attempt_count
    attempt_count += 1
    if attempt_count < 3:
        raise ValueError("not yet")
    return "ok"

result = flaky()
print(result, attempt_count)   # should print: ok 3

Exercise 3 — Add @functools.wraps:

Take the shout decorator from Exercise 1 and add @functools.wraps(func).

Then run print(add.__name__) with and without it. What changes?

Exercise 4 — Write a context manager:

Write a context manager called log_step(name) that:

• Prints `"Starting: {name}"` on entry
• Prints `"Done: {name} ({elapsed:.1f}s)"` on exit
• Prints `"Failed: {name} — {error}"` if an exception is raised

(and re-raises the exception after printing)

Test it by wrapping your qwen_call in a scratch script.

INTERMEDIATE — Lesson 09

Lesson 09 — Async and Concurrency

Source files: scripts/daily-content-agent.py, scripts/trendpilot-loop.py

---

The Problem: 14 Sequential API Calls

daily-content-agent.py writes 14 articles per day (2 per product × 7 products).

Each Qwen API call takes roughly 10–30 seconds. They run one after the other:

for product in products:
    for keyword in pick_keywords(product, day_index, count=2):
        write_article(product, keyword, day_index, article_num)

That's up to 7 minutes of wall-clock time where most of your program is just

waiting for the network. Your CPU sits idle while bytes trickle in.

The same pattern appears in trendpilot-loop.py where it calls Qwen for product

descriptions, trend analysis, and strategy — all sequential.

With async, you could fire all 14 calls at once and wait for them all to finish.

Estimated improvement: 7 minutes → under 1 minute.

---

Two Different Problems, Two Different Tools

Before reaching for async, understand what you're waiting on:

Waiting for	Tool	Why
-------------	------	-----
Network I/O (API calls, HTTP)	`asyncio`	The wait is in the kernel, CPU is free
File I/O on slow disk	`asyncio` with `aiofiles`	Same idea
CPU-heavy computation	`multiprocessing`	GIL blocks threads for CPU work
Blocking library code	`threading`	Releases the GIL during I/O

Your scripts are waiting on network I/O. That's the asyncio case.

---

What Async Actually Means

Normally, your program runs instructions one at a time:

Start Qwen call 1 ──▶ wait 15s ──▶ receive ──▶ write file
Start Qwen call 2 ──▶ wait 12s ──▶ receive ──▶ write file
...
Total: 15 + 12 + ... = 105 seconds

With async, Python can switch tasks while waiting:

Start Qwen call 1 ──▶ (waiting...) ──────────────────▶ receive ──▶ write file
Start Qwen call 2     ──▶ (waiting...) ────────────▶ receive ──▶ write file
Start Qwen call 3         ──▶ (waiting...) ──────────────▶ receive ──▶ write
...
Total: max(15, 12, ...) ≈ 30 seconds (longest single call)

The key word is waiting. Async only helps when your program is blocked waiting

for something external (network, disk). If you're crunching numbers, async won't help.

---

asyncio Basics

import asyncio

# Mark a function as async with "async def"
async def say_hello():
    print("Hello")
    await asyncio.sleep(1)   # "await" suspends here, lets other tasks run
    print("World")

# Run it
asyncio.run(say_hello())

await is the pause point. While this coroutine is suspended at await,

the event loop can run other coroutines.

You can only use await inside async def functions.

---

Async Version of qwen_call

Your current qwen_call uses urllib.request.urlopen which blocks the entire thread.

To use it with asyncio, switch to aiohttp (a third-party library):

pip install aiohttp

import aiohttp, asyncio, json

async def qwen_call_async(session: aiohttp.ClientSession,
                           prompt: str,
                           model: str = "qwen-plus",
                           temperature: float = 0.7,
                           max_tokens: int = 4000) -> str:
    """Async version of qwen_call. Pass a shared session for connection reuse."""
    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "max_tokens": max_tokens,
        "stream": False,
    }
    async with session.post(QWEN_API_URL, json=payload, headers=headers,
                             timeout=aiohttp.ClientTimeout(total=120)) as resp:
        resp.raise_for_status()   # raises on 4xx/5xx
        data = await resp.json()
        return data["choices"][0]["message"]["content"]

Key differences from the sync version:

• `async def` instead of `def`
• `aiohttp.ClientSession` instead of `urllib.request`
• `await resp.json()` instead of `json.loads(resp.read())`
• The session is shared (connection reuse = faster)

---

Running 14 Calls Concurrently

import asyncio
import aiohttp

async def write_article_async(session, product, keyword, day_index, article_num):
    slug = slugify(keyword)
    file_path = BLOG_DIR / f"{product}-{slug}.md"
    prompt = build_prompt(product, keyword)

    try:
        article = await qwen_call_async(session, prompt)
        file_path.write_text(article)
        print(f"  DONE {product}/{keyword[:40]}")
        return str(file_path)
    except Exception as e:
        print(f"  ERROR {product}/{keyword[:40]}: {e}")
        file_path.write_text(f"# {keyword.title()}\n\n*Writing...*\n")
        return str(file_path)


async def main_async():
    products = ["resumeforge", "clara", "clozr", "localeye", "agentseek", "agentpay", "cleaneats"]
    day_index = get_day_index()

    tasks = []
    async with aiohttp.ClientSession() as session:
        for product in products:
            for i, keyword in enumerate(pick_keywords(product, day_index, count=2)):
                task = write_article_async(session, product, keyword, day_index, i + 1)
                tasks.append(task)

        # Fire all 14 tasks at once, wait for all to finish
        results = await asyncio.gather(*tasks, return_exceptions=True)

    return [r for r in results if isinstance(r, str)]


if __name__ == "__main__":
    urls = asyncio.run(main_async())
    submit_to_gsc(urls)

asyncio.gather(*tasks) starts all tasks concurrently and returns when all finish.

return_exceptions=True means a failed task returns the exception instead of

crashing everything — important when you have 14 tasks and one might hit a rate limit.

---

Rate Limits: Don't Be Too Aggressive

Sending 14 requests simultaneously might trigger Qwen's rate limiter (429).

Add a semaphore to cap concurrent requests:

async def main_async():
    semaphore = asyncio.Semaphore(3)   # max 3 concurrent Qwen calls

    async def limited_write(session, product, keyword, day_index, article_num):
        async with semaphore:          # blocks if 3 are already running
            return await write_article_async(session, product, keyword, day_index, article_num)

    async with aiohttp.ClientSession() as session:
        tasks = [
            limited_write(session, product, keyword, day_index, i + 1)
            for product in products
            for i, keyword in enumerate(pick_keywords(product, day_index, count=2))
        ]
        return await asyncio.gather(*tasks, return_exceptions=True)

With Semaphore(3), at most 3 calls run at once. 14 calls become roughly 5 batches

of 3 — still about 5x faster than sequential.

---

Threading: The Alternative

If you don't want to rewrite your functions as async, threading gives you

concurrency with your existing sync code. HTTP calls release the GIL during

network waits, so threads actually help here:

from concurrent.futures import ThreadPoolExecutor, as_completed

def main_threaded():
    products = ["resumeforge", "clara", "clozr", "localeye", "agentseek", "agentpay", "cleaneats"]
    day_index = get_day_index()

    tasks = [
        (product, keyword, day_index, i + 1)
        for product in products
        for i, keyword in enumerate(pick_keywords(product, day_index, count=2))
    ]

    results = []
    with ThreadPoolExecutor(max_workers=3) as pool:
        futures = {
            pool.submit(write_article, product, keyword, day_index, n): (product, keyword)
            for product, keyword, day_index, n in tasks
        }
        for future in as_completed(futures):
            product, keyword = futures[future]
            try:
                results.append(future.result())
            except Exception as e:
                print(f"  ERROR {product}/{keyword}: {e}")

    return results

ThreadPoolExecutor is easier to add to existing sync code — no need to change

function signatures. The tradeoff: threads use more memory than async coroutines.

Rule of thumb: If you're starting fresh, use asyncio. If you're adding

concurrency to existing sync code, start with ThreadPoolExecutor.

---

When NOT to Use Async

Don't add async complexity for scripts that run one-off tasks quickly. Async pays

off when:

• You have 5+ concurrent API calls
• You're building a server that handles multiple requests simultaneously
• Wall-clock time matters (automated pipelines, not interactive scripts)

For a script you run manually once, sequential is fine and much simpler to debug.

---

Exercises

Exercise 1 — Understand the event loop:

Run this and predict the output before you do:

import asyncio

async def task(name, delay):
    print(f"Start {name}")
    await asyncio.sleep(delay)
    print(f"Done  {name}")

async def main():
    await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )

asyncio.run(main())

What order does "Done" print? Why?

Exercise 2 — Time the difference:

Write two versions of "fetch N URLs" — one sequential with urllib.request, one

concurrent with ThreadPoolExecutor(max_workers=5). Use 10 calls to

https://httpbin.org/delay/1 (each takes 1 second). Time both with time.perf_counter().

You should see roughly 10s vs 2s.

Exercise 3 — Add a semaphore:

In the main_async function above, change Semaphore(3) to Semaphore(1).

What does that make the concurrent version equivalent to? What's the point of

asyncio.gather at that point?

Exercise 4 — Add concurrency to daily-content-agent.py:

Add ThreadPoolExecutor(max_workers=3) to the main() function in

daily-content-agent.py so all 14 write_article calls run concurrently

(up to 3 at a time). Don't change write_article itself. Time how long it

takes before and after.

INTERMEDIATE — Lesson 10

Lesson 10 — Testing and Deployment

Source files: amazon-price-tracker/main.py, scripts/daily-content-agent.py

---

The Problem: Zero Tests, Zero Safety Net

You have scripts running in production — writing articles, posting to Facebook,

managing Shopify inventory — and zero automated tests. That means:

• Every bug is discovered in production
• You can't refactor with confidence ("what if I broke something?")
• Debugging means reading logs and guessing

amazon-price-tracker/main.py is your most structured script. It has classes,

type hints, real database code. It's also completely untested. If check_price_drops

has an off-by-one error in its percentage calculation, you'd only find out when you

stopped getting alerts.

---

What's Worth Testing

Not everything needs a test. The rule: test pure functions (same input → same

output, no network, no files). Don't test code that's mostly network calls — test the

logic around those calls.

In amazon-price-tracker/main.py, these are worth testing:

# Pure logic — easy to test
def should_send_alert(current_price: float, target_price: float) -> bool:
    return current_price <= target_price

# Pure logic — easy to test
def calculate_savings(original: float, current: float) -> float:
    return original - current

def calculate_savings_pct(original: float, current: float) -> float:
    if original == 0:
        return 0.0
    return ((original - current) / original) * 100

These don't hit the network or database. You can run them in milliseconds.

# Harder — touches SQLite, file system
def setup_database(self):
    conn = sqlite3.connect(self.db_path)
    ...

# Hard — network call to Amazon
def fetch_amazon_price(self, asin: str):
    ...

For the database and network code, you test the *behavior* by mocking the call.

---

pytest Basics

Install: pip install pytest

Create a file: tests/test_price_logic.py

# tests/test_price_logic.py

def should_send_alert(current_price: float, target_price: float) -> bool:
    return current_price <= target_price

def calculate_savings_pct(original: float, current: float) -> float:
    if original == 0:
        return 0.0
    return ((original - current) / original) * 100


# Tests — any function named test_* is auto-discovered by pytest
def test_alert_triggers_when_below_target():
    assert should_send_alert(current_price=4.50, target_price=5.00) is True

def test_alert_does_not_trigger_when_above_target():
    assert should_send_alert(current_price=6.00, target_price=5.00) is False

def test_alert_triggers_at_exact_target():
    assert should_send_alert(current_price=5.00, target_price=5.00) is True

def test_savings_pct_normal():
    pct = calculate_savings_pct(original=10.00, current=7.50)
    assert pct == 25.0

def test_savings_pct_zero_original():
    # Guard against division by zero
    pct = calculate_savings_pct(original=0.0, current=5.00)
    assert pct == 0.0

def test_savings_pct_no_savings():
    pct = calculate_savings_pct(original=10.00, current=10.00)
    assert pct == 0.0

Run: pytest tests/

Output:

collected 6 items

tests/test_price_logic.py ......     [100%]

6 passed in 0.03s

Each . is a passing test. A F is a failure — pytest shows you exactly what

the expected vs actual values were.

---

Testing with Mocks: Fake the Network

unittest.mock lets you replace a real function with a fake one during a test.

Use this to test code that normally makes network calls:

# tests/test_price_monitor.py
from unittest.mock import patch, MagicMock
import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from amazon-price-tracker.main import AmazonPriceMonitor

@patch("main.requests.get")   # replace requests.get with a fake
def test_fetch_price_returns_float(mock_get):
    # Set up what the fake should return
    mock_response = MagicMock()
    mock_response.json.return_value = {
        "data": {"product": {"current_price": "9.99"}}
    }
    mock_response.status_code = 200
    mock_get.return_value = mock_response

    monitor = AmazonPriceMonitor.__new__(AmazonPriceMonitor)
    monitor.base_url = "https://real-time-amazon-data.p.rapidapi.com"
    monitor.api_key = "fake-key"

    price = monitor.fetch_amazon_price("B08H95Y452")
    assert price == 9.99
    assert isinstance(price, float)

@patch("main.requests.get") intercepts the requests.get call and returns your

mock_response instead. The network never gets called. The test runs in milliseconds.

---

Testing Your Content Agent Logic

daily-content-agent.py has pure functions worth testing immediately:

# tests/test_content_agent.py
import sys
sys.path.insert(0, "scripts")
from daily_content_agent import slugify, pick_keywords

def test_slugify_basic():
    assert slugify("how to write a resume") == "how-to-write-a-resume"

def test_slugify_truncates_at_60():
    long = "a" * 100
    assert len(slugify(long)) <= 60

def test_slugify_lowercases():
    assert slugify("UPPER CASE") == "upper-case"

def test_pick_keywords_returns_correct_count():
    keywords = pick_keywords("resumeforge", day_index=0, count=2)
    assert len(keywords) == 2

def test_pick_keywords_rotates_by_day():
    # Day 0 and day 15 should start at different positions
    kw_day0 = pick_keywords("resumeforge", day_index=0, count=1)
    kw_day15 = pick_keywords("resumeforge", day_index=15, count=1)
    # They might overlap (wrapping), but the logic should rotate
    assert isinstance(kw_day0[0], str)
    assert isinstance(kw_day15[0], str)

def test_pick_keywords_unknown_product():
    # Unknown product returns empty list
    keywords = pick_keywords("nonexistent_product", day_index=0, count=2)
    assert keywords == []

Run: pytest tests/test_content_agent.py -v

The -v flag shows each test name and pass/fail individually.

---

Test Structure: Arrange, Act, Assert

Every test follows the same three-step pattern:

def test_alert_triggers_when_below_target():
    # Arrange — set up inputs
    current_price = 4.50
    target_price = 5.00

    # Act — call the code being tested
    result = should_send_alert(current_price, target_price)

    # Assert — check the result
    assert result is True

Keep each test focused on one behavior. If a test fails, the name should tell

you exactly what broke.

---

Running Tests Automatically: CI and Pre-commit

A test you only run when you remember is not much better than no test.

Option 1: Run before every commit with a shell alias

# In ~/.bashrc or ~/.zshrc
alias commit='pytest tests/ && git commit'

Option 2: GitHub Actions (runs on every push)

Create .github/workflows/test.yml:

name: Tests
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install pytest aiohttp
      - run: pytest tests/ -v

Every time you push to GitHub, this runs. If a test fails, GitHub marks the commit

red and sends you an email.

---

Cron Deployment

Your scripts already use cron. Here's how the full deployment lifecycle works for

a script like daily-content-agent.py:

View current cron jobs:

crontab -l

Add a new cron job:

crontab -e

Cron syntax: minute hour day month weekday command

# Run daily-content-agent.py at 6:00 AM every day
0 6 * * * /usr/bin/python3 /home/ron/.openclaw/workspace/scripts/daily-content-agent.py >> /home/ron/.openclaw/workspace/logs/daily-content.log 2>&1

# Run price tracker every 5 minutes
*/5 * * * * /usr/bin/python3 /home/ron/.openclaw/workspace/amazon-price-tracker/main.py >> /home/ron/.openclaw/workspace/logs/price-tracker.log 2>&1

>> logfile 2>&1 redirects both stdout and stderr to the log file, appending.

Always use absolute paths in cron — cron doesn't have your PATH or current

directory. python3 might not be found; use /usr/bin/python3 or the full path

from which python3.

Check if cron is running what you think:

grep CRON /var/log/syslog | tail -20

---

Structuring Code for Testability

The key insight: code that's hard to test is usually hard to understand and maintain.

Testability is a proxy for good design.

The biggest improvement you can make to daily-content-agent.py:

# Before: logic mixed with side effects — hard to test
def write_article(product, keyword, day_index, article_num):
    slug = slugify(keyword)
    file_path = BLOG_DIR / f"{product}-{slug}.md"
    prompt = f"""Write an SEO blog article..."""  # built inline
    article = qwen_call(prompt)                    # network call
    if isinstance(article, dict) and "error" in article:
        ...
    file_path.write_text(article)                  # file write
    html_path.write_text(md_to_html(...))          # another file write
    return str(file_path)

# After: pure functions separated from side effects
def build_article_prompt(product: str, keyword: str) -> str:
    """Pure — no network, no files. Easy to test."""
    ...

def parse_article_response(raw: str) -> str:
    """Pure — validates and cleans the response. Easy to test."""
    ...

def write_article(product, keyword, day_index, article_num):
    """Thin orchestrator — hard to test, but thin means less can go wrong."""
    prompt = build_article_prompt(product, keyword)    # testable
    article = qwen_call(prompt)                        # external — mock in tests
    cleaned = parse_article_response(article)          # testable
    file_path.write_text(cleaned)                      # side effect

Now build_article_prompt and parse_article_response are just functions that

take strings and return strings. You can test 20 cases in 20ms without touching

the network or file system.

---

Exercises

Exercise 1 — Your first test file:

Create tests/test_slugify.py. Import slugify from daily-content-agent.py

and write 4 tests:

• Spaces become hyphens
• Output is lowercase
• Output is at most 60 chars
• Double spaces don't become double hyphens (`"a b"` → `"a-b"`, not `"a--b"`)

Run pytest tests/test_slugify.py -v. Fix any failures you find — they might

reveal an actual bug.

Exercise 2 — Test pick_keywords rotation:

Write a test that verifies pick_keywords with day_index=0 returns different

keywords than day_index=1 (for the same product and count). If this test fails,

the rotation logic is broken.

Exercise 3 — Mock the API:

In daily-content-agent.py, find a function that calls qwen_call and does

something with the result. Write a test using unittest.mock.patch that:

• Patches `qwen_call` to return a fixed string (no network call)
• Calls the outer function
• Asserts the output file contains the expected content

Exercise 4 — Add tests to CI:

Push your tests/ folder to GitHub. Create .github/workflows/test.yml using

the template above. Make a commit and watch the Actions tab — does the green

checkmark appear?