186 lines
6.5 KiB
Python
186 lines
6.5 KiB
Python
"""One-shot import of FabledRulebook .md files into the Scribe DB.
|
|
|
|
Usage:
|
|
python -m scripts.import_fabledrulebook \\
|
|
--source-dir /path/to/FabledRulebook \\
|
|
--user-id 1 \\
|
|
[--rulebook-title "FabledSword family"] \\
|
|
[--subscribe-projects 3,4,5] \\
|
|
[--dry-run] \\
|
|
[--force]
|
|
|
|
Calls the services layer (services/rulebooks.py) — does not write raw SQL.
|
|
This means any structural mismatch surfaces as a clean service-level
|
|
ValueError. The script is not part of Alembic; the operator runs it once
|
|
after migration 0055 lands.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
RULE_HEADING_RE = re.compile(r"^##\s+(.+?)\s*$", re.MULTILINE)
|
|
WHY_RE = re.compile(r"\*\*\s*Why:?\s*\*\*", re.IGNORECASE)
|
|
HOW_RE = re.compile(r"\*\*\s*How to apply:?\s*\*\*", re.IGNORECASE)
|
|
H1_RE = re.compile(r"^#\s+.+$", re.MULTILINE)
|
|
|
|
|
|
logger = logging.getLogger("import_fabledrulebook")
|
|
|
|
|
|
def parse_rule_body(title: str, body: str) -> dict:
|
|
"""Extract statement, why, how_to_apply from a rule's H2-section body."""
|
|
body = body.strip()
|
|
statement = body
|
|
why = ""
|
|
how_to_apply = ""
|
|
|
|
why_match = WHY_RE.search(body)
|
|
how_match = HOW_RE.search(body)
|
|
|
|
if why_match:
|
|
statement = body[: why_match.start()].strip()
|
|
if how_match and how_match.start() > why_match.end():
|
|
why = body[why_match.end() : how_match.start()].strip()
|
|
how_to_apply = body[how_match.end() :].strip()
|
|
else:
|
|
why = body[why_match.end() :].strip()
|
|
elif how_match:
|
|
statement = body[: how_match.start()].strip()
|
|
how_to_apply = body[how_match.end() :].strip()
|
|
|
|
return {
|
|
"title": title,
|
|
"statement": statement,
|
|
"why": why,
|
|
"how_to_apply": how_to_apply,
|
|
}
|
|
|
|
|
|
def parse_topic_file(path: Path) -> tuple[str, str, list[dict]]:
|
|
"""Returns (topic_title, topic_description, [rules])."""
|
|
text = path.read_text(encoding="utf-8")
|
|
topic_title = path.stem # e.g. "git-workflow"
|
|
|
|
parts = RULE_HEADING_RE.split(text)
|
|
preamble = parts[0] if parts else ""
|
|
# Strip H1 from preamble to get just the description.
|
|
description = H1_RE.sub("", preamble, count=1).strip()
|
|
|
|
rules: list[dict] = []
|
|
for i in range(1, len(parts), 2):
|
|
rule_title = parts[i].strip()
|
|
body = parts[i + 1] if i + 1 < len(parts) else ""
|
|
rules.append(parse_rule_body(rule_title, body))
|
|
|
|
return topic_title, description, rules
|
|
|
|
|
|
async def run(args) -> None:
|
|
source = Path(args.source_dir)
|
|
if not source.exists():
|
|
raise SystemExit(f"Source dir not found: {source}")
|
|
|
|
md_files = sorted(p for p in source.glob("*.md") if p.name.lower() != "readme.md")
|
|
if not md_files:
|
|
raise SystemExit(f"No topic .md files found in {source}")
|
|
|
|
readme = source / "README.md"
|
|
description = readme.read_text(encoding="utf-8") if readme.exists() else ""
|
|
|
|
if args.dry_run:
|
|
print(f"DRY RUN — would import {len(md_files)} topic file(s):")
|
|
for f in md_files:
|
|
topic_title, _, rules = parse_topic_file(f)
|
|
print(f" TOPIC: {topic_title} ({len(rules)} rules)")
|
|
for r in rules:
|
|
preview = r["statement"][:80] + ("…" if len(r["statement"]) > 80 else "")
|
|
empty_flag = " [EMPTY STATEMENT — will skip]" if not r["statement"].strip() else ""
|
|
print(f" - {r['title']}: {preview}{empty_flag}")
|
|
return
|
|
|
|
from fabledassistant.services import rulebooks as rb_service
|
|
|
|
existing = await rb_service.find_rulebook_by_title(args.user_id, args.rulebook_title)
|
|
if existing is not None and not args.force:
|
|
raise SystemExit(
|
|
f"Rulebook '{args.rulebook_title}' already exists for user "
|
|
f"{args.user_id}; pass --force to create another."
|
|
)
|
|
|
|
rb = await rb_service.create_rulebook(
|
|
user_id=args.user_id,
|
|
title=args.rulebook_title,
|
|
description=description,
|
|
)
|
|
logger.info("Created rulebook %d ('%s')", rb.id, rb.title)
|
|
|
|
rules_created = 0
|
|
rules_skipped = 0
|
|
for i, f in enumerate(md_files):
|
|
topic_title, topic_desc, rules = parse_topic_file(f)
|
|
topic = await rb_service.create_topic(
|
|
rulebook_id=rb.id,
|
|
user_id=args.user_id,
|
|
title=topic_title,
|
|
description=topic_desc,
|
|
order_index=i,
|
|
)
|
|
logger.info(" Created topic %d ('%s', %d rules)", topic.id, topic.title, len(rules))
|
|
for j, rule in enumerate(rules):
|
|
if not rule["statement"].strip():
|
|
logger.warning(" SKIP empty-statement rule: %s / %s", topic_title, rule["title"])
|
|
rules_skipped += 1
|
|
continue
|
|
await rb_service.create_rule(
|
|
topic_id=topic.id,
|
|
user_id=args.user_id,
|
|
title=rule["title"],
|
|
statement=rule["statement"],
|
|
why=rule["why"],
|
|
how_to_apply=rule["how_to_apply"],
|
|
order_index=j,
|
|
)
|
|
rules_created += 1
|
|
|
|
for pid in (args.subscribe_projects or []):
|
|
await rb_service.subscribe_project(
|
|
project_id=pid, rulebook_id=rb.id, user_id=args.user_id,
|
|
)
|
|
logger.info("Subscribed project %d to rulebook %d", pid, rb.id)
|
|
|
|
print(
|
|
f"\nImported rulebook '{rb.title}' (id={rb.id}): "
|
|
f"{len(md_files)} topics, {rules_created} rules created"
|
|
+ (f", {rules_skipped} skipped" if rules_skipped else "")
|
|
)
|
|
|
|
|
|
def _comma_ints(s: str) -> list[int]:
|
|
return [int(x) for x in s.split(",") if x.strip()]
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source-dir", required=True,
|
|
help="Path to the FabledRulebook directory")
|
|
parser.add_argument("--user-id", type=int, required=True,
|
|
help="Scribe user id who will own the rulebook")
|
|
parser.add_argument("--rulebook-title", default="FabledSword family")
|
|
parser.add_argument("--subscribe-projects", type=_comma_ints,
|
|
help="Comma-separated project ids to auto-subscribe")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
parser.add_argument("--force", action="store_true",
|
|
help="Allow creating a rulebook even if title exists")
|
|
args = parser.parse_args()
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
|
asyncio.run(run(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|