Codex スキルで X アカウント運用を自動化する方法

your-project/
├── .codex/
│   └── skills/
│       └── x-growth-operator/
│           ├── SKILL.md
│           ├── templates/
│           │   └── account-brief.yaml
│           └── scripts/
│               ├── mix_ledger.py
│               ├── hook_lint.py
│               ├── metrics.py
│               └── weekly_review.py
└── x-ops/
    ├── account-brief.yaml        ← you fill this in, once
    ├── mix-ledger.tsv            ← header row only
    ├── metrics.tsv               ← header row only
    └── scripts/                  ← copy of the skill's scripts/

printf 'date	slot	shape	target_signal	hook_pattern	topic	post_url
' > x-ops/mix-ledger.tsv
printf 'post_url	collected_at	impressions	likes	replies	reposts	bookmarks	profile_clicks
' > x-ops/metrics.tsv

---
name: x-growth-operator
description: Run the operating loop for one X account. Chooses today's content slot from a logged mix, drafts the post in the matching shape, rejects weak hooks with a deterministic gate, records what was published, and reviews performance from an X API pull or a manually imported metrics file. Use when the user asks to plan, draft, gate, or review posts for their own X account.
---

# X Growth Operator

One account per `x-ops/` directory. The state files are the memory; read them rather than working from recollection.

```text
x-ops/
├── account-brief.yaml    positioning, pillars, tone, mix target, signal weights
├── mix-ledger.tsv        one row per published post
├── metrics.tsv           one row per post per collection date
└── scripts/
    ├── mix_ledger.py     plan / log / report
    ├── hook_lint.py      the four-check hook gate
    ├── metrics.py        pull (X API) / import (your own export)
    └── weekly_review.py  join, medians, named patterns
```

## 0. Rules that never bend

1. Never post, schedule, or call any X write endpoint. Output is a draft block for the user to copy. Publishing is a human action.
2. Never state a metric that is not in `x-ops/metrics.tsv`. Missing data is reported as missing, never estimated.
3. Never recommend a slot without running `mix_ledger.py plan` first.
4. Never produce engagement pods, bulk follow or unfollow sequences, or the same text repeated across accounts. If asked, refuse and name the platform rule it violates.
5. Never draft a post whose only purpose is to trigger a reaction the content does not support.

## 1. First run

If `x-ops/account-brief.yaml` is missing or still the template, stop and build it with the user before anything else. Ask, do not infer. The `mix_target` block is required by `mix_ledger.py`; the rest guides your drafting.

Copy `scripts/` alongside the skill and create the two state files as header-only tables:

```bash
cp -r <skill-dir>/scripts x-ops/scripts
printf 'date	slot	shape	target_signal	hook_pattern	topic	post_url
' > x-ops/mix-ledger.tsv
printf 'post_url	collected_at	impressions	likes	replies	reposts	bookmarks	profile_clicks
' > x-ops/metrics.tsv
```

Then write `x-ops/account-brief.yaml` from `templates/account-brief.yaml` with the user, filling in every field.

## 2. The daily loop

Four steps, in order. Do not skip the gate because a draft looks good.

### 2.1 Plan

```bash
python3 x-ops/scripts/mix_ledger.py plan --window 20
```

This reads the trailing window, compares the counts per slot against `mix_target`, and prints the slot with the largest deficit plus the arithmetic behind the pick. Report its output. Do not substitute your own judgement about which slot would be more interesting.

When the window holds fewer than 20 rows, the script says so. Pass that warning on rather than hiding it.

### 2.2 Draft

Load the shape for the selected slot from §9 and write the post.

- Single post: hook, the core claim in one or two sentences, the evidence, exactly one call to action.
- Thread: see §3.
- The target signal drives the ending. A post aimed at bookmarks ends with something worth keeping. A post aimed at replies ends with something a reader can disagree with.

### 2.3 Hook gate

```bash
python3 x-ops/scripts/hook_lint.py --hook "<first line>" --body "<full post>"
```

The script runs four mechanical checks and exits non-zero on any failure:

1. Banned openers, meaning openers that would still be true of a different post.
2. Concreteness: the hook needs a number, a named thing, or a scene. Otherwise it is a thesis wearing a hook's clothing.
3. Gap width: a total claim with no figure or boundary is too wide to close.
4. Payoff: every number and name in the hook must appear in the body.

The gate is four questions, not an oracle. Checks 2 and 3 are heuristics and will occasionally reject a deliberately plain hook. When you disagree with a rejection, keep the hook and say why. When you agree, show the rejected line and the rewrite side by side; never present only the fixed version.

A gate that never rejects anything is not running. If ten consecutive drafts all pass, read §9's hook failure table with the user and add their own clichés to the `BANNED` list in `hook_lint.py`.

### 2.4 Log

After the user confirms they published, append the row:

```bash
python3 x-ops/scripts/mix_ledger.py log \
  --slot A --shape tutorial --signal bookmark \
  --hook-pattern information-gap --topic "short topic label" \
  --url "x.com/HANDLE/status/ID"
```

Ask before logging; do not assume publication. A row with no url is unverified and cannot be joined to metrics later.

## 3. Thread mode

Use a thread when the slot is A or B and the material genuinely needs steps. Do not use a thread to say one thing at length.

At a target of five slots:

| Slot | Job | Fails when |
| --- | --- | --- |
| 1 | Promise the outcome, not the topic | It describes the subject instead of the result |
| 2-4 | The two to four points that carry the argument | More than one idea per slot |
| 5-6 | Steps, checklist, or a worked case | It restates slot 1 instead of adding proof |
| Last | Summary plus one call to action | Two calls to action |

Five is a target. Three is fine when the material is thin; seven only if every slot does distinct work.

## 4. Reply engine

Replies carry the second-highest weight in the planning model, which makes writing one a distribution action rather than a courtesy. Run it as its own pass.

Build a target list from accounts the user follows inside their topic pillars, and for each produce a reply in one of four kinds:

1. Add a number the original post left out.
2. Offer a counterexample from the user's own work, without hostility.
3. Ask the follow-up question the post's own logic demands.
4. Describe what happened when the user tried it.

Reject bare praise, agreement with no addition, replies that are really an advertisement for the user's own post, and any reply that drags in an unrelated account for reach. Do not use the mention mechanism on people who have not asked for it.

## 5. The two metrics paths

Both write the same file. The review does not care which one produced a row.

### Path A, the X API

```bash
export X_BEARER_TOKEN="..."          # user-context OAuth 2.0 token
python3 x-ops/scripts/metrics.py pull --dry-run     # inspect the request first
python3 x-ops/scripts/metrics.py pull
```

The script reads post ids from the ledger, batches them 100 at a time, requests both the public and non-public metric groups, and writes only the counters that come back. Bookmark and impression counts require user-context authentication and are returned only for the authenticated account's own posts, which is exactly the scope this loop needs.

Field availability, tier names, prices, and rate limits on the X developer platform change often. Confirm the current ones against the developer documentation before building anything on top, and never answer a rate-limit rejection by guessing the numbers.

### Path B, manual import

```bash
python3 x-ops/scripts/metrics.py import --file export.csv
python3 x-ops/scripts/metrics.py import --file export.csv --map "url=Post link,impressions=Views"
```

The script recognizes common header names automatically and leaves anything it cannot map as an empty field. An empty field is honest and is reported as unavailable. A filled-in guess contaminates every review built on it afterward.

Never write a metrics row to test the review. Run the review against an empty table instead; it should say it has no data rather than produce a summary.

## 6. Weekly review

```bash
python3 x-ops/scripts/weekly_review.py --window 20
```

The script joins the ledger to the metrics on the post id, excludes rows collected inside the 48-hour window, reports medians per slot, and names the patterns it can prove:

- High impressions, low reposts: reached people, did not travel.
- High bookmarks, low impressions: worth keeping, under-distributed. Rewrite around a new angle; do not re-post it.
- Low replies across every slot: a shape problem, not a post problem.
- One slot outperforming the rest on reach: record it and keep the ratio.

Report what it prints, including the gap lists. When the data cannot support a claim, say so instead of summarizing around the hole.

## 7. Failure modes

| Symptom | Likely cause | Recovery |
| --- | --- | --- |
| Plan returns the same slot repeatedly | Window below 20 rows, or posts are not being logged | Confirm the log step runs. Say the window is thin instead of repeating a slot |
| Every draft passes the gate | The gate is being read, not applied | Add the user's own clichés to `BANNED` and re-run |
| Metrics fail to join | Post url formats differ between ledger and export | Normalize both to the numeric post id |
| Review dominated by one post | A mean crept in, or the window is too small | Medians are already used; widen the window and report the outlier |
| Mix drifts anyway | Posts are published outside the loop | Log them. Off-loop posts are still part of the mix |
| Drafts sound different each week | The brief has no banned list | Add specific banned moves, each with an example |

## 8. The 48-hour window

The planning model treats the first two days after publication as the period when a post can still be picked up. Two consequences.

Do not re-run a post from outside the window as though it were fresh. If it deserves reviving, rewrite it around a new angle and publish new material.

Do not judge a post before the window closes. `weekly_review.py` excludes the early rows automatically; do not override that by hand.

## 9. Reference tables

### The four account roles

Pick exactly one for the brief. Everything else assumes a single consistent position.

| Role | The account is | Publishes mostly |
| --- | --- | --- |
| Research translator | Reading primary sources and restating them clearly | Slot B |
| Engineering practitioner | Building things and reporting what broke | Slot A |
| Product and application advocate | Showing what a tool does in real use | Slots A and C |
| Industry observer | Tracking where the field is going and betting on it | Slots B and D |

### Content mix

| Slot | Share | What it is |
| --- | --- | --- |
| A | 50 | Hands-on tutorials, prompts, and worked skill or tool examples |
| B | 20 | Close reads of papers, releases, and trend shifts |
| C | 15 | Project progress and retrospectives |
| D | 10 | Industry opinion with a stated position |
| E | 5 | The flexible slot, for whatever the week hands you |

### The six shapes

| Shape | Opens with | Aimed at |
| --- | --- | --- |
| Discovery | A thing the reader did not know existed | Bookmark, profile click |
| Review | I tested this, here is the verdict | Reply, bookmark |
| Tutorial | Here is how to do the thing | Bookmark |
| Opinion | Here is what I think, and why | Reply |
| Retro | Here is what happened and what I learned | Reply, profile click |
| Announcement | Here is what shipped | Profile click, repost |

### Three hook tension sources

1. **Information gap.** Withhold one specific thing the reader needs, and make the edges of the gap visible. A gap the reader cannot see the shape of reads as vagueness.
2. **Scene resonance.** Open inside a situation the reader has been in. A scene earns the second sentence; an abstraction does not.
3. **Counterintuitive conflict.** State the thing that contradicts expectation, then earn it. Only usable when the body actually earns it.

### Four hooks that fail

| Failure | What it looks like | Fix |
| --- | --- | --- |
| Thesis in a hook's clothing | The main claim, stated flat, with no gap and no scene | Move the claim into the body. Open with the evidence that made you believe it |
| Template that outlived its edits | An opener that fits any post on any account | Delete the first line. Start with the second |
| Gap opened too wide | A total claim with no figure, boundary, or condition | Add the constraint. Say what it does not apply to |
| Gap the body never closes | A sharp hook followed by general advice | Cut the hook, or write the body the hook promised |

### Single post structure

Hook, then the core claim in one or two sentences, then the evidence, then one call to action. Two to three sentences total for the common case.

### The verified output block

Every draft leaves the loop in this shape:

```text
SLOT      A (deficit 2.0 over a 20-row window)
SHAPE     tutorial
SIGNAL    bookmark
HOOK      <the first line>
BODY      <the post>
CTA       <the one action>
GATE      check1 pass | check2 pass | check3 pass | check4 pass
```

If the GATE line is missing, the draft did not go through the gate. Run it again before showing the draft.

# x-ops/account-brief.yaml
# Fill this in with the user before the first planning run. Every field here is
# read by the skill or by a script. Do not leave a field as a guess.

account:
  handle: ""
  positioning: ""        # exactly one of the four roles in SKILL.md section 9
  audience: ""           # who reads this, in one sentence
  language: ""           # the language post drafts are written in

pillars:                 # 2-4 topics this account is allowed to publish about
  - ""
  - ""

tone:
  allowed: []            # moves that fit this account
  banned: []             # words or moves it never uses, each with a reason

# Read by mix_ledger.py, which parses this block only. Must total 100.
mix_target:
  A: 50
  B: 20
  C: 15
  D: 10
  E: 5

# Planning weights, not measurements. These drive which signal a draft is aimed
# at, not how results are scored. The top three are the relative multipliers
# this skill was tuned around. The middle band is ordinal, meaning "more than a
# like, less than a bookmark", and the numbers are editable placeholders.
# Change any of them and say so in the review.
signal_weights:
  repost: 20
  reply: 13.5
  bookmark: 10
  profile_click: 8
  link_click: 7
  video_completion: 6
  dwell: 6
  like: 1

thread_length: 5         # target slots, usable range 3 to 7
window: 20               # the trailing post count every ratio is computed over

#!/usr/bin/env python3
"""Content-mix ledger for the x-growth-operator skill.

One job: never let a slot be chosen from memory. The ledger is the record,
this script is the arithmetic.

Usage:
  mix_ledger.py plan   [--window 20] [--ops DIR]
  mix_ledger.py log    --slot A --shape tutorial --signal bookmark
                       --hook-pattern information-gap --topic "..."
                       [--url URL] [--date YYYY-MM-DD] [--ops DIR]
  mix_ledger.py report [--window 20] [--ops DIR]

Reads the mix target from x-ops/account-brief.yaml (the mix_target block only)
and the record from x-ops/mix-ledger.tsv.
"""

import argparse
import datetime as dt
import os
import re
import sys

HEADER = ["date", "slot", "shape", "target_signal", "hook_pattern", "topic", "post_url"]


def ops_dir(arg):
    return arg or os.environ.get("X_OPS_DIR") or "x-ops"


def read_target(brief_path):
    """Return {slot: share} from the mix_target block, or None if unreadable."""
    if not os.path.exists(brief_path):
        return None
    target = {}
    inside = False
    with open(brief_path, encoding="utf-8") as fh:
        for raw in fh:
            line = raw.rstrip("
")
            if re.match(r"^mix_target\s*:", line):
                inside = True
                continue
            if inside:
                if line.strip() and not line.startswith((" ", "	")):
                    break  # next top-level key
                m = re.match(r"^\s+([A-Za-z])\s*:\s*(\d+(?:\.\d+)?)\s*$", line)
                if m:
                    target[m.group(1).upper()] = float(m.group(2))
    return target or None


def read_rows(path):
    if not os.path.exists(path):
        return []
    rows = []
    with open(path, encoding="utf-8") as fh:
        for n, raw in enumerate(fh):
            line = raw.rstrip("
")
            if not line.strip():
                continue
            parts = line.split("	")
            if n == 0 and parts[0].strip().lower() == "date":
                continue  # header
            if parts[0].strip().lower() == "date":
                continue
            parts += [""] * (len(HEADER) - len(parts))
            rows.append(dict(zip(HEADER, parts)))
    return rows


def window_rows(rows, size):
    return rows[-size:] if size and size > 0 else rows


def deficits(rows, target):
    """Positive number means the slot is behind its share of the window."""
    n = len(rows)
    order = sorted(target.keys())
    counts = {s: 0 for s in order}
    for r in rows:
        s = (r["slot"] or "").strip().upper()
        if s in counts:
            counts[s] += 1
    total = sum(target.values()) or 100.0
    out = {}
    for s in order:
        share = target[s] / total
        out[s] = {
            "count": counts[s],
            "expected": share * n,
            "deficit": share * n - counts[s],
            "share_actual": (counts[s] / n * 100) if n else 0.0,
            "share_target": share * 100,
        }
    return out, counts, n


def last_seen(rows):
    seen = {}
    for i, r in enumerate(rows):
        s = (r["slot"] or "").strip().upper()
        if s:
            seen[s] = (r["date"], i)
    return seen


def cmd_plan(args):
    d = ops_dir(args.ops)
    ledger = os.path.join(d, "mix-ledger.tsv")
    target = read_target(os.path.join(d, "account-brief.yaml"))
    if not target:
        sys.exit("could not read a mix_target block from %s/account-brief.yaml" % d)

    rows = window_rows(read_rows(ledger), args.window)
    info, counts, n = deficits(rows, target)

    if n == 0:
        pick = sorted(target.keys())[0]
        print("window: empty, no rows logged yet")
        print("slot:   %s (default first slot; the ledger has nothing to count)" % pick)
        print("note:   below %d rows the deficit is noise. Publish and log anyway." % args.window)
        return 0

    thin = n < args.window
    seen = last_seen(rows)
    ranked = sorted(
        info.items(),
        key=lambda kv: (-kv[1]["deficit"], seen.get(kv[0], ("", -1))[1]),
    )
    pick, stats = ranked[0]

    print("window: last %d logged posts%s" % (n, " (THIN, fewer than %d)" % args.window if thin else ""))
    print()
    print("slot  count  expected  deficit  actual%   target%")
    for s, st in sorted(info.items()):
        print(
            "  %-4s %5d %9.1f %8.1f %7.1f%% %7.1f%%"
            % (s, st["count"], st["expected"], st["deficit"], st["share_actual"], st["share_target"])
        )
    print()
    print("slot:   %s" % pick)
    print("why:    deficit %.1f over a %d-row window%s" % (stats["deficit"], n, " (tie broken by least recent)" if len(ranked) > 1 and ranked[1][1]["deficit"] == stats["deficit"] else ""))
    if thin:
        print("warning: window below %d rows. Treat the pick as provisional." % args.window)
    return 0


def cmd_log(args):
    d = ops_dir(args.ops)
    os.makedirs(d, exist_ok=True)
    ledger = os.path.join(d, "mix-ledger.tsv")
    new = not os.path.exists(ledger)
    date = args.date or dt.date.today().isoformat()
    row = [date, args.slot.upper(), args.shape, args.signal, args.hook_pattern, args.topic, args.url or ""]
    if "	" in "".join(row):
        sys.exit("fields may not contain tabs")
    with open(ledger, "a", encoding="utf-8") as fh:
        if new:
            fh.write("	".join(HEADER) + "
")
        fh.write("	".join(row) + "
")
    print("logged: %s" % "	".join(row))
    if not args.url:
        print("note:   post_url is empty, so this row is unverified and cannot be joined to metrics")
    return 0


def cmd_report(args):
    d = ops_dir(args.ops)
    target = read_target(os.path.join(d, "account-brief.yaml"))
    if not target:
        sys.exit("could not read a mix_target block from %s/account-brief.yaml" % d)
    rows = window_rows(read_rows(os.path.join(d, "mix-ledger.tsv")), args.window)
    info, counts, n = deficits(rows, target)
    if n == 0:
        print("no rows in the window; nothing to report")
        return 0
    print("%d rows in window" % n)
    print()
    print("slot  count  expected  delta   actual%   target%   status")
    for s, st in sorted(info.items()):
        status = "ok"
        if st["deficit"] > 1.5:
            status = "BEHIND"
        elif st["deficit"] < -1.5:
            status = "over"
        print(
            "  %-4s %5d %9.1f %+7.1f %7.1f%% %7.1f%%  %s"
            % (s, st["count"], st["expected"], st["deficit"], st["share_actual"], st["share_target"], status)
        )
    return 0


def main():
    ap = argparse.ArgumentParser(description="x-growth-operator content-mix ledger")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("plan", help="pick today's slot from the trailing window")
    p.add_argument("--window", type=int, default=20)
    p.add_argument("--ops")
    p.set_defaults(func=cmd_plan)

    p = sub.add_parser("log", help="append a published post to the ledger")
    p.add_argument("--slot", required=True)
    p.add_argument("--shape", required=True)
    p.add_argument("--signal", required=True, help="the signal this post targets")
    p.add_argument("--hook-pattern", required=True, dest="hook_pattern")
    p.add_argument("--topic", required=True)
    p.add_argument("--url", default="")
    p.add_argument("--date")
    p.add_argument("--ops")
    p.set_defaults(func=cmd_log)

    p = sub.add_parser("report", help="actual mix vs target")
    p.add_argument("--window", type=int, default=20)
    p.add_argument("--ops")
    p.set_defaults(func=cmd_report)

    args = ap.parse_args()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Deterministic hook gate for the x-growth-operator skill.

Four checks on the first line of a draft, all mechanical. Exits 1 if any
check fails so a caller can branch on the exit code.

Usage:
  hook_lint.py --hook "..." --body "..."
  hook_lint.py --hook "..." --body-file post.txt
  hook_lint.py --card x-ops/post-card.md          # reads HOOK/BODY blocks

Checks
  1 banned openers      - openers that fit any post on any account
  2 concreteness        - needs a number, a name, or a scene
  3 gap width           - total claims with no figure or boundary
  4 payoff              - every number and name in the hook appears in the body

Check 2 and 3 are heuristics that catch the two most common failure shapes.
They produce false positives on deliberately plain hooks. When you disagree
with a rejection, keep the hook and say why; the gate is four questions, not
an oracle.
"""

import argparse
import re
import sys

BANNED = [
    r"^in today'?s",
    r"^let'?s dive in",
    r"^let'?s talk about",
    r"^we'?re excited to",
    r"^excited to announce",
    r"^here'?s why",
    r"^here are \d+ (things|ways|tips|reasons)",
    r"^a thread",
    r"^thread",
    r"^hot take",
    r"^unpopular opinion",
    r"^game[- ]?chang(er|ing)",
    r"^the future of \w+ is",
    r"^\w+ is (dead|dying)",
    r"^stop doing",
    r"^nobody talks about",
    r"^most people (get|don'?t)",
]

# Words that signal a total claim the body probably cannot pay off.
WIDE = [
    "everything", "everyone", "nobody", "always", "never", "all of",
    "the only", "completely", "totally", "100%", "forever", "instantly",
]

SCENE_MARKERS = [
    r"i", r"we", r"my", r"our", r"yesterday",
    r"last (week|month|night)", r"this (morning|week|month)",
    r"when i", r"after \d", r"three (weeks|months|days)",
]

# Capitalised words that are not proper nouns.
STOP_CAPS = {
    "The", "This", "That", "These", "Those", "It", "We", "I", "You", "They",
    "He", "She", "A", "An", "And", "But", "Or", "So", "If", "When", "Why",
    "How", "What", "Every", "Most", "Some", "No", "Not", "Do", "Does", "Did",
    "My", "Our", "Your", "Their", "Its", "There", "Here", "Stop", "Start",
    "Never", "Always", "Just", "Only", "One", "Two", "Three", "In", "On",
    "At", "After", "Before", "Because", "Then", "Now", "Today", "Yesterday",
    "Everything", "Nothing", "Everyone", "Nobody", "Anything", "Something",
    "Anyone", "Someone", "Everywhere", "Nothing's", "Its",
}

CAP = re.compile(r"[A-Z][A-Za-z0-9.+#-]{1,}")
SENT_SPLIT = re.compile(r"(?<=[.!?])\s+")


def proper_nouns(text):
    """Capitalised words that are not the first token of a sentence.

    Sentence-initial capitals carry no signal, which is the whole reason this
    is not a one-line regex.
    """
    found = []
    for sentence in SENT_SPLIT.split(text.strip()):
        tokens = sentence.split()
        for tok in tokens[1:]:
            for word in CAP.findall(tok):
                if word not in STOP_CAPS:
                    found.append(word)
    return found


def first_line(hook):
    return hook.strip().splitlines()[0].strip() if hook.strip() else ""


def check_banned(hook):
    low = hook.lower()
    for pat in BANNED:
        if re.search(pat, low):
            return False, "banned opener matching /%s/" % pat
    return True, "no banned opener"


def check_concrete(hook):
    digits = re.findall(r"\d", hook)
    caps = proper_nouns(hook)
    scene = [p for p in SCENE_MARKERS if re.search(p, hook, re.I)]
    if digits:
        return True, "carries a number"
    if caps:
        return True, "names %s" % ", ".join(sorted(set(caps))[:3])
    if scene:
        return True, "opens inside a scene"
    return False, "no number, no name, no scene. This reads as a thesis, not a hook"


def check_width(hook):
    low = hook.lower()
    hits = [w for w in WIDE if re.search(r"%s" % re.escape(w), low)]
    has_number = bool(re.search(r"\d", hook))
    if hits and not has_number:
        return False, "total claim (%s) with no figure or boundary to close it" % ", ".join(hits)
    if hits:
        return True, "total claim present but bounded by a figure"
    return True, "gap width not obviously too wide"


def check_payoff(hook, body):
    tokens = set(re.findall(r"\d+(?:[.,]\d+)?", hook))
    tokens |= set(proper_nouns(hook))
    missing = sorted(t for t in tokens if t not in body)
    if missing:
        return False, "the body never mentions %s" % ", ".join(missing)
    return True, "every number and name in the hook appears in the body"


def run(hook, body):
    line = first_line(hook)
    if not line:
        print("FAIL  the hook is empty")
        return 1
    results = [
        ("1 banned openers", check_banned(line)),
        ("2 concreteness  ", check_concrete(line)),
        ("3 gap width     ", check_width(line)),
        ("4 payoff        ", check_payoff(line, body or "")),
    ]
    print("hook: %s" % line)
    print("chars: %d" % len(line))
    print()
    failed = 0
    for name, (ok, why) in results:
        print("%s  %s  %s" % ("pass" if ok else "FAIL", name, why))
        if not ok:
            failed += 1
    print()
    if failed:
        print("gate: REJECTED on %d check(s). Rewrite the hook, do not soften the body." % failed)
        return 1
    print("gate: passed")
    return 0


def read_card(path):
    text = open(path, encoding="utf-8").read()
    hook = body = ""
    m = re.search(r"^HOOK[ 	]*(.*)$", text, re.M)
    if m:
        hook = m.group(1).strip()
    m = re.search(r"^BODY[ 	]*(.*?)(?=^\w+[ 	]|\Z)", text, re.M | re.S)
    if m:
        body = m.group(1).strip()
    return hook, body


def main():
    ap = argparse.ArgumentParser(description="x-growth-operator hook gate")
    ap.add_argument("--hook")
    ap.add_argument("--body", default="")
    ap.add_argument("--body-file")
    ap.add_argument("--card")
    args = ap.parse_args()

    hook, body = args.hook, args.body
    if args.card:
        hook, body = read_card(args.card)
    if args.body_file:
        body = open(args.body_file, encoding="utf-8").read()
    if not hook:
        sys.exit("no hook supplied; pass --hook or --card")
    return run(hook, body)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Metrics ingestion for the x-growth-operator skill.

Two paths, one output file. The review does not care which path produced a row.

  metrics.py pull    read post ids from the ledger, ask the X API, append rows
  metrics.py import  read your own CSV/TSV, map columns, append rows

Both append to x-ops/metrics.tsv:
  post_url  collected_at  impressions  likes  replies  reposts  bookmarks  profile_clicks

An absent counter is written as an empty field and reported as unavailable.
Nothing is ever estimated.

Path A needs a user-context OAuth 2.0 token in $X_BEARER_TOKEN, and an access
tier that permits reading. Field availability differs by tier and auth method,
so this script requests both the public and non-public metric groups and fills
only the counters that come back. Confirm the current tier, fields, and rate
limits against X's developer documentation before relying on them.
"""

import argparse
import csv
import datetime as dt
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

HEADER = ["post_url", "collected_at", "impressions", "likes", "replies", "reposts", "bookmarks", "profile_clicks"]

API = "https://api.x.com/2/tweets"
BATCH = 100

# API field group -> our column
FIELD_MAP = {
    "impression_count": "impressions",
    "like_count": "likes",
    "reply_count": "replies",
    "retweet_count": "reposts",
    "bookmark_count": "bookmarks",
    "user_profile_clicks": "profile_clicks",
}

# Accepted header names per column, for the manual path.
ALIASES = {
    "post_url": ["post_url", "url", "post url", "link", "permalink", "tweet url", "post link"],
    "collected_at": ["collected_at", "date", "collected", "export date"],
    "impressions": ["impressions", "impression_count", "impressions total", "views"],
    "likes": ["likes", "like_count", "favorites", "likes total"],
    "replies": ["replies", "reply_count", "replies total"],
    "reposts": ["reposts", "retweets", "retweet_count", "reposts total"],
    "bookmarks": ["bookmarks", "bookmark_count", "bookmarks total"],
    "profile_clicks": ["profile_clicks", "user_profile_clicks", "profile clicks", "profile visits"],
}


def ops_dir(arg):
    return arg or os.environ.get("X_OPS_DIR") or "x-ops"


def append_rows(path, rows):
    new = not os.path.exists(path)
    with open(path, "a", encoding="utf-8", newline="") as fh:
        w = csv.writer(fh, delimiter="	")
        if new:
            w.writerow(HEADER)
        for r in rows:
            w.writerow([r.get(c, "") for c in HEADER])


def post_ids_from_ledger(path):
    ids, order = [], []
    if not os.path.exists(path):
        sys.exit("no ledger at %s" % path)
    with open(path, encoding="utf-8") as fh:
        for raw in fh:
            parts = raw.rstrip("
").split("	")
            if len(parts) < 7 or parts[0].strip().lower() == "date":
                continue
            url = parts[6].strip()
            if not url or url.startswith("<"):
                continue
            m = re.search(r"/status(?:es)?/(\d+)", url) or re.fullmatch(r"(\d+)", url)
            if m:
                ids.append(m.group(1))
                order.append(url)
    return ids, order


def api_get(url, token, timeout=30):
    req = urllib.request.Request(url, headers={
        "Authorization": "Bearer %s" % token,
        "User-Agent": "x-growth-operator metrics.py",
    })
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode("utf-8"))


def cmd_pull(args):
    d = ops_dir(args.ops)
    ids, order = post_ids_from_ledger(os.path.join(d, "mix-ledger.tsv"))
    if not ids:
        print("no post ids in the ledger. Log a published post with --url first.")
        return 0

    collected = args.collected or dt.date.today().isoformat()
    fields = "public_metrics,non_public_metrics,created_at"
    batches = [ids[i:i + BATCH] for i in range(0, len(ids), BATCH)]

    if args.dry_run:
        for b in batches:
            print("%s?ids=%s&tweet.fields=%s" % (API, ",".join(b), fields))
        print("
%d ids in %d request(s). Token read from $X_BEARER_TOKEN." % (len(ids), len(batches)))
        return 0

    token = os.environ.get("X_BEARER_TOKEN")
    if not token:
        sys.exit("$X_BEARER_TOKEN is not set. Export a user-context OAuth 2.0 token, or use the import path.")

    rows, missing = [], []
    for b in batches:
        q = urllib.parse.urlencode({"ids": ",".join(b), "tweet.fields": fields})
        url = "%s?%s" % (API, q)
        for attempt in range(4):
            try:
                payload = api_get(url, token)
                break
            except urllib.error.HTTPError as e:
                if e.code == 429 and attempt < 3:
                    wait = int(e.headers.get("x-rate-limit-reset", "0"))
                    delay = max(5, min(60, wait - int(time.time()))) if wait else 15
                    print("rate limited, waiting %ds" % delay, file=sys.stderr)
                    time.sleep(delay)
                    continue
                print("HTTP %s on batch starting %s: %s" % (e.code, b[0], e.read()[:200].decode("utf-8", "replace")), file=sys.stderr)
                payload = None
                break
            except Exception as e:
                print("request failed: %s" % e, file=sys.stderr)
                payload = None
                break
        if not payload:
            missing.extend(b)
            continue

        seen = set()
        for item in payload.get("data") or []:
            seen.add(item.get("id"))
            values = {"post_url": "x.com/i/status/%s" % item["id"], "collected_at": collected}
            groups = [item.get("public_metrics") or {}, item.get("non_public_metrics") or {}]
            for src_col, dest in FIELD_MAP.items():
                for g in groups:
                    if src_col in g:
                        values[dest] = g[src_col]
                        break
            # keep the original url when we can match it back
            for u in order:
                if u.endswith(str(item.get("id"))):
                    values["post_url"] = u
                    break
            rows.append(values)

        if payload.get("errors"):
            for err in payload["errors"]:
                rid = (err.get("value") or err.get("resource_id") or "?")
                missing.append(str(rid))
        missing.extend([i for i in b if i not in seen and i not in missing])

    if rows:
        append_rows(os.path.join(d, "metrics.tsv"), rows)
    print("appended %d row(s) for %s" % (len(rows), collected))
    if missing:
        print("no data returned for %d id(s): %s" % (len(set(missing)), ", ".join(sorted(set(missing))[:8])))
    if not rows:
        print("nothing written. Check that the token is user-context and the tier allows reads.")
    return 0


def detect_delim(sample):
    return "	" if sample.count("	") > sample.count(",") else ","


def cmd_import(args):
    d = ops_dir(args.ops)
    path = args.file
    if path == "-":
        sample = sys.stdin.read()
        fh = sample.splitlines()
    else:
        with open(path, encoding="utf-8") as f:
            sample = f.read()
        fh = sample.splitlines()
    if not fh:
        sys.exit("empty input")

    delim = detect_delim(sample[:4000])
    reader = csv.DictReader(fh, delimiter=delim)
    headers = reader.fieldnames or []

    mapping = {}
    explicit = {}
    if args.map:
        for pair in args.map.split(","):
            if "=" in pair:
                k, v = pair.split("=", 1)
                explicit[k.strip().lower()] = v.strip()

    for col in HEADER:
        if col in explicit:
            src = explicit[col]
            if src and src not in headers:
                sys.exit("--map points %s at %r, which is not in the file. Headers: %s" % (col, src, headers))
            mapping[col] = src
            continue
        lower = {h.lower().strip(): h for h in headers}
        for alias in ALIASES[col]:
            if alias in lower:
                mapping[col] = lower[alias]
                break
        else:
            mapping[col] = None

    unmapped = [c for c in HEADER if not mapping.get(c) and c != "collected_at"]
    if unmapped:
        print("unmapped columns, will be left empty: %s" % ", ".join(unmapped))
        print("use --map \"%s=<your header>\" to fill them" % unmapped[0])

    collected = args.collected or dt.date.today().isoformat()
    rows, blanks = [], {c: 0 for c in HEADER}
    for rec in reader:
        row = {}
        for col in HEADER:
            src = mapping.get(col)
            val = ""
            if src:
                val = (rec.get(src) or "").strip()
            if col == "collected_at" and not val:
                val = collected
            if col == "post_url" and re.fullmatch(r"\d+", val or ""):
                # a bare post id becomes a url; anything else is left alone
                val = "x.com/i/status/%s" % val
            if not val and col != "post_url":
                blanks[col] += 1
            row[col] = val
        if row["post_url"]:
            rows.append(row)

    if not rows:
        sys.exit("no usable rows: every row was missing a post url")

    append_rows(os.path.join(d, "metrics.tsv"), rows)
    print("appended %d row(s) for %s" % (len(rows), collected))
    thin = [c for c, n in blanks.items() if n == len(rows) and c != "post_url"]
    if thin:
        print("always empty, reported as unavailable rather than zero: %s" % ", ".join(thin))
    return 0


def main():
    ap = argparse.ArgumentParser(description="x-growth-operator metrics ingestion")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("pull", help="path A: pull from the X API")
    p.add_argument("--collected")
    p.add_argument("--ops")
    p.add_argument("--dry-run", action="store_true", dest="dry_run")
    p.set_defaults(func=cmd_pull)

    p = sub.add_parser("import", help="path B: import your own export")
    p.add_argument("--file", required=True, help="CSV/TSV path, or - for stdin")
    p.add_argument("--map", help='override column mapping, e.g. "url=Post link,impressions=Views"')
    p.add_argument("--collected")
    p.add_argument("--ops")
    p.set_defaults(func=cmd_import)

    args = ap.parse_args()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Weekly review for the x-growth-operator skill.

Joins the ledger to the metrics, reports per slot letter, and names the four
patterns worth acting on. Medians, not means: one good post should not define
a whole slot.

Rows collected inside the 48-hour window are excluded by default, because a
post at hour six has not finished.

Usage:
  weekly_review.py [--window 20] [--exclude-hours 48] [--min-posts 3] [--ops DIR]
"""

import argparse
import csv
import datetime as dt
import os
import statistics as st
import sys

COUNTERS = ["impressions", "likes", "replies", "reposts", "bookmarks", "profile_clicks"]
LEDGER_HEADER = ["date", "slot", "shape", "target_signal", "hook_pattern", "topic", "post_url"]


def ops_dir(arg):
    return arg or os.environ.get("X_OPS_DIR") or "x-ops"


def read_tsv(path, header):
    if not os.path.exists(path):
        return []
    rows = []
    with open(path, encoding="utf-8") as fh:
        rdr = csv.reader(fh, delimiter="	")
        for i, parts in enumerate(rdr):
            if not parts or not any(p.strip() for p in parts):
                continue
            if i == 0 and parts[0].strip().lower() in ("date", "post_url"):
                continue
            if parts[0].strip().lower() in ("date", "post_url"):
                continue
            parts = parts + [""] * (len(header) - len(parts))
            rows.append(dict(zip(header, parts)))
    return rows


def key(url):
    """Normalise a post url to a bare id so the two files can be joined."""
    url = (url or "").strip().split("?")[0].rstrip("/")
    return url.rsplit("/", 1)[-1].lower() if url else ""


def parse_date(s):
    s = (s or "").strip()
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y", "%d/%m/%Y"):
        try:
            return dt.datetime.strptime(s, fmt).date()
        except ValueError:
            continue
    return None


def to_num(v):
    v = (v or "").strip().replace(",", "").replace("%", "")
    if not v:
        return None
    try:
        return float(v)
    except ValueError:
        return None


def main():
    ap = argparse.ArgumentParser(description="x-growth-operator weekly review")
    ap.add_argument("--window", type=int, default=20)
    ap.add_argument("--exclude-hours", type=int, default=48, dest="exclude_hours")
    ap.add_argument("--min-posts", type=int, default=3, dest="min_posts",
                    help="minimum posts in a slot before its median is reported")
    ap.add_argument("--ops")
    args = ap.parse_args()

    d = ops_dir(args.ops)
    ledger = read_tsv(os.path.join(d, "mix-ledger.tsv"), LEDGER_HEADER)
    metrics = read_tsv(os.path.join(d, "metrics.tsv"), ["post_url", "collected_at"] + COUNTERS)

    if not ledger:
        print("no ledger rows. Nothing to review.")
        return 0

    ledger = ledger[-args.window:] if args.window > 0 else ledger
    by_id = {}
    for r in metrics:
        k = key(r["post_url"])
        if not k:
            continue
        by_id.setdefault(k, []).append(r)

    joined, no_metrics, early = [], [], 0
    for r in ledger:
        k = key(r["post_url"])
        if not k:
            no_metrics.append(r)
            continue
        candidates = by_id.get(k)
        if not candidates:
            no_metrics.append(r)
            continue
        pub = parse_date(r["date"])
        usable = []
        for c in candidates:
            cd = parse_date(c["collected_at"])
            if pub and cd and (cd - pub).days * 24 < args.exclude_hours:
                early += 1
                continue
            usable.append(c)
        if not usable:
            no_metrics.append(r)
            continue
        latest = max(usable, key=lambda c: parse_date(c["collected_at"]) or dt.date.min)
        joined.append((r, latest))

    orphans = [r for r in metrics if key(r["post_url"]) not in {key(l["post_url"]) for l in ledger}]

    print("# Weekly review")
    print()
    print("%d ledger rows in window, %d joined to metrics, %d without usable metrics."
          % (len(ledger), len(joined), len(no_metrics)))
    if early:
        print("%d metrics row(s) excluded for being inside the %dh window."
              % (early, args.exclude_hours))
    print()

    if not joined:
        print("No joinable data. The review cannot say anything yet, and it will not guess.")
        if no_metrics:
            no_url = [r for r in no_metrics if not (r.get("post_url") or "").strip()]
            waiting = [r for r in no_metrics if (r.get("post_url") or "").strip()]
            print()
            if no_url:
                print("Ledger rows with no url (%d). These can never join:" % len(no_url))
                for r in no_url[:10]:
                    print("  %s  %s  %s" % (r["date"], r["slot"], r["topic"] or "(no topic)"))
            if waiting:
                print("Rows whose metrics are missing or still inside the %dh window (%d):"
                      % (args.exclude_hours, len(waiting)))
                for r in waiting[:10]:
                    print("  %s  %s  %s" % (r["date"], r["slot"], r["topic"] or "(no topic)"))
                print("Ingest again once the window closes. Lowering --exclude-hours to see numbers early is the failure this rule exists to prevent.")
        return 0

    slots = {}
    for l, m in joined:
        slots.setdefault((l["slot"] or "?").upper(), []).append((l, m))

    print("## By slot")
    print()
    print("slot  n  " + "  ".join("%s" % c.rjust(14) for c in COUNTERS))
    for s in sorted(slots):
        items = slots[s]
        cells = []
        for c in COUNTERS:
            vals = [to_num(m.get(c)) for _, m in items]
            vals = [v for v in vals if v is not None]
            if len(vals) < args.min_posts:
                cells.append("n/a".rjust(14))
            else:
                cells.append(("%.0f" % st.median(vals)).rjust(14))
        print("%-5s %d  %s" % (s, len(items), "  ".join(cells)))
    print()
    print("n/a means fewer than %d posts with that counter. It does not mean zero." % args.min_posts)
    print()

    print("## Patterns")
    print()
    flags = 0
    allr = [(l, m) for l, m in joined]
    imp = [(l, to_num(m.get("impressions"))) for l, m in allr if to_num(m.get("impressions")) is not None]
    if len(imp) >= args.min_posts:
        med_imp = st.median([v for _, v in imp])
        rep = [(l, to_num(m.get("reposts"))) for l, m in allr if to_num(m.get("reposts")) is not None]
        bkm = [(l, to_num(m.get("bookmarks"))) for l, m in allr if to_num(m.get("bookmarks")) is not None]
        if rep:
            med_rep = st.median([v for _, v in rep])
            hits = []
            for l, m in allr:
                i, rp = to_num(m.get("impressions")), to_num(m.get("reposts"))
                if i is not None and rp is not None and i > med_imp * 1.3 and rp <= med_rep:
                    hits.append(l)
            if hits:
                flags += 1
                print("**High impressions, low reposts** on %d post(s): reached people, did not travel." % len(hits))
                for l in hits[:5]:
                    print("  %s  %s  %s" % (l["date"], l["slot"], l["topic"]))
                print()
        if bkm:
            med_bk = st.median([v for _, v in bkm])
            hits = []
            for l, m in allr:
                i, bk = to_num(m.get("impressions")), to_num(m.get("bookmarks"))
                if i is not None and bk is not None and bk > med_bk * 1.3 and i <= med_imp:
                    hits.append(l)
            if hits:
                flags += 1
                print("**High bookmarks, low impressions** on %d post(s): worth keeping, nobody saw it." % len(hits))
                print("  Rewrite around a new angle rather than re-posting it.")
                for l in hits[:5]:
                    print("  %s  %s  %s" % (l["date"], l["slot"], l["topic"]))
                print()

    reps = [to_num(m.get("replies")) for _, m in allr]
    reps = [v for v in reps if v is not None]
    if reps:
        med_r = st.median(reps)
        per_slot = {}
        for s, items in slots.items():
            vals = [to_num(m.get("replies")) for _, m in items]
            vals = [v for v in vals if v is not None]
            if vals:
                per_slot[s] = st.median(vals)
        if per_slot and all(v <= med_r for v in per_slot.values()) and med_r < 3:
            flags += 1
            print("**Low replies across every slot** (median %.1f). The account is not inviting" % med_r)
            print("  disagreement anywhere. That is a shape problem, not a post problem.")
            print()

    if len(slots) >= 2:
        med_by_slot = {}
        for s, items in slots.items():
            vals = [to_num(m.get("impressions")) for _, m in items]
            vals = [v for v in vals if v is not None]
            if len(vals) >= args.min_posts:
                med_by_slot[s] = st.median(vals)
        if len(med_by_slot) >= 2 and max(med_by_slot.values()) > 1.6 * min(med_by_slot.values()):
            flags += 1
            best = max(med_by_slot, key=med_by_slot.get)
            print("**One slot outperforming the rest**: %s at %.0f median impressions vs %.0f for %s."
                  % (best, med_by_slot[best], min(med_by_slot.values()), min(med_by_slot, key=med_by_slot.get)))
            print("  Record it and keep the ratio. Reach is not the assignment. Change the brief")
            print("  only if the gap holds across two full windows.")
            print()

    if not flags:
        print("No patterns above threshold this window. That is a valid result, not a bug.")
        print()

    print("## Gaps")
    print()
    if no_metrics:
        print("- %d ledger row(s) with no usable metrics." % len(no_metrics))
        for r in no_metrics[:5]:
            print("    %s  %s  %s" % (r["date"], r["slot"], r["topic"] or "(no topic)"))
    if orphans:
        print("- %d metrics row(s) with no ledger entry. Publish outside the loop still counts." % len(orphans))
        for r in orphans[:5]:
            print("    %s  %s" % (r["collected_at"], r["post_url"]))
    if not no_metrics and not orphans:
        print("- none. Every row joined in both directions.")
    print()
    print("Every number above traces to a row in x-ops/metrics.tsv.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

$ python3 x-ops/scripts/mix_ledger.py plan --window 20
window: last 20 logged posts

slot  count  expected  deficit  actual%   target%
  A       14      10.0     -4.0    70.0%    50.0%
  B        4       4.0      0.0    20.0%    20.0%
  C        1       3.0      2.0     5.0%    15.0%
  D        1       2.0      1.0     5.0%    10.0%
  E        0       1.0      1.0     0.0%     5.0%

slot:   C
why:    deficit 2.0 over a 20-row window

$ python3 x-ops/scripts/hook_lint.py \
    --hook "In today's fast-paced AI landscape, building an audience is everything." \
    --body "Some general advice about posting."
hook: In today's fast-paced AI landscape, building an audience is everything.
chars: 71

FAIL  1 banned openers  banned opener matching /^in today'?s/
pass  2 concreteness    names AI
FAIL  3 gap width       total claim (everything) with no figure or boundary to close it
FAIL  4 payoff          the body never mentions AI

gate: REJECTED on 3 check(s). Rewrite the hook, do not soften the body.

$ python3 x-ops/scripts/hook_lint.py \
    --hook "I ran 40 posts through the same gate. 31 failed on one check." \
    --body "I ran 40 posts through the same gate last month. 31 failed on one check: the body never paid off the hook."
hook: I ran 40 posts through the same gate. 31 failed on one check.
chars: 61

pass  1 banned openers  no banned opener
pass  2 concreteness    carries a number
pass  3 gap width       gap width not obviously too wide
pass  4 payoff          every number and name in the hook appears in the body

gate: passed

python3 x-ops/scripts/mix_ledger.py log \
  --slot C --shape retro --signal reply \
  --hook-pattern scene --topic "gate rejection rate" \
  --url "x.com/yourhandle/status/1234567890"

export X_BEARER_TOKEN="your user-context OAuth 2.0 token"
python3 x-ops/scripts/metrics.py pull --dry-run   # prints the request, calls nothing
python3 x-ops/scripts/metrics.py pull

python3 x-ops/scripts/metrics.py import --file export.csv
python3 x-ops/scripts/metrics.py import --file export.csv --map "url=Post link,impressions=Views"

python3 x-ops/scripts/weekly_review.py --window 20