7850baf6bb
Dump stackexchange_20260331 (newest mirror), 6 sites, downloads running; streamed ingest/filter/tags pipeline; status writer + homepage strip show the layer's build progress every 5 minutes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PDKeXvpT6tENSvyQGLV1Uq
278 lines
10 KiB
Python
278 lines
10 KiB
Python
"""SkillFactor knowledge layer — Stack Exchange practitioner Q&A.
|
|
|
|
Reproducible pipeline over the archive.org data-dump mirror
|
|
(stackexchange_20260331, CC-BY-SA 4.0). See knowledge/README.md.
|
|
|
|
CLI:
|
|
python knowledge/pipeline/se_pipeline.py ingest 7z -> SQLite (streamed)
|
|
python knowledge/pipeline/se_pipeline.py filter quality gates
|
|
python knowledge/pipeline/se_pipeline.py tags tag frequency export
|
|
python knowledge/pipeline/se_pipeline.py compile profession knowledge files
|
|
python knowledge/pipeline/se_pipeline.py verify consistency checks
|
|
|
|
Idempotent: existing archives are not re-downloaded (scripts/download.ps1),
|
|
extraction/ingest skip completed sites, summaries are cached per post id.
|
|
Every step updates data/progress-se.json for the homepage live status.
|
|
"""
|
|
import argparse
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime
|
|
from xml.etree import ElementTree as ET
|
|
|
|
K = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # knowledge/
|
|
RAW = os.path.join(K, "data", "raw")
|
|
EXTRACTED = os.path.join(K, "data", "extracted")
|
|
DB = os.path.join(K, "data", "se.db")
|
|
PROGRESS = os.path.join(K, "data", "progress-se.json")
|
|
|
|
DUMP_VERSION = "stackexchange_20260331"
|
|
DUMP_URL = "https://archive.org/download/stackexchange_20260331/"
|
|
|
|
SITES = {
|
|
"workplace": "workplace.stackexchange.com",
|
|
"pm": "pm.stackexchange.com",
|
|
"law": "law.stackexchange.com",
|
|
"money": "money.stackexchange.com",
|
|
"softwareengineering": "softwareengineering.stackexchange.com",
|
|
"datascience": "datascience.stackexchange.com",
|
|
}
|
|
|
|
MIN_Q_SCORE = 5
|
|
MIN_A_SCORE = 5
|
|
MIN_ANSWER_CHARS = 300
|
|
TAG_MIN_FREQ = 20
|
|
|
|
|
|
def progress(step, **info):
|
|
data = {}
|
|
if os.path.exists(PROGRESS):
|
|
try:
|
|
data = json.load(open(PROGRESS, encoding="utf-8"))
|
|
except ValueError:
|
|
data = {}
|
|
data["step"] = step
|
|
data["updated_at"] = datetime.now().isoformat(timespec="seconds")
|
|
data.setdefault("steps", {})[step] = {
|
|
"at": data["updated_at"], **info}
|
|
json.dump(data, open(PROGRESS, "w", encoding="utf-8"), indent=1)
|
|
|
|
|
|
def strip_html(s):
|
|
if not s:
|
|
return ""
|
|
s = re.sub(r"<(pre|code)[^>]*>(.*?)</\1>",
|
|
lambda m: "\n```\n" + m.group(2) + "\n```\n", s, flags=re.S)
|
|
s = re.sub(r"<br\s*/?>|</p>|</li>", "\n", s)
|
|
s = re.sub(r"<[^>]+>", "", s)
|
|
return html.unescape(s).strip()
|
|
|
|
|
|
def db():
|
|
cn = sqlite3.connect(DB)
|
|
cn.execute("PRAGMA journal_mode=WAL")
|
|
return cn
|
|
|
|
|
|
def ensure_schema(cn):
|
|
cn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS questions(
|
|
site TEXT, id INTEGER, title TEXT, body TEXT, tags TEXT,
|
|
score INTEGER, accepted_answer_id INTEGER, view_count INTEGER,
|
|
closed INTEGER, owner_id INTEGER, PRIMARY KEY(site, id));
|
|
CREATE TABLE IF NOT EXISTS answers(
|
|
site TEXT, id INTEGER, parent_id INTEGER, body TEXT,
|
|
score INTEGER, owner_id INTEGER, PRIMARY KEY(site, id));
|
|
CREATE TABLE IF NOT EXISTS users(
|
|
site TEXT, id INTEGER, display_name TEXT, PRIMARY KEY(site, id));
|
|
CREATE TABLE IF NOT EXISTS ingest_log(
|
|
site TEXT PRIMARY KEY, questions INTEGER, answers INTEGER,
|
|
users INTEGER, at TEXT);
|
|
CREATE INDEX IF NOT EXISTS ix_answers_parent ON answers(site, parent_id);
|
|
""")
|
|
|
|
|
|
def cmd_ingest():
|
|
import py7zr
|
|
cn = db()
|
|
ensure_schema(cn)
|
|
done = {r[0] for r in cn.execute("SELECT site FROM ingest_log")}
|
|
for key, host in SITES.items():
|
|
if key in done:
|
|
print(f"{key}: already ingested — skip")
|
|
continue
|
|
arc = os.path.join(RAW, f"{host}.7z")
|
|
if not os.path.exists(arc):
|
|
print(f"{key}: archive missing ({arc}) — skip")
|
|
continue
|
|
xdir = os.path.join(EXTRACTED, key)
|
|
os.makedirs(xdir, exist_ok=True)
|
|
if not os.path.exists(os.path.join(xdir, "Posts.xml")):
|
|
print(f"{key}: extracting Posts/Tags/Users ...")
|
|
with py7zr.SevenZipFile(arc) as z:
|
|
names = [n for n in z.getnames()
|
|
if n in ("Posts.xml", "Tags.xml", "Users.xml")]
|
|
z.extract(path=xdir, targets=names)
|
|
|
|
nq = na = nu = 0
|
|
qb, ab = [], []
|
|
for _ev, el in ET.iterparse(os.path.join(xdir, "Posts.xml")):
|
|
if el.tag != "row":
|
|
continue
|
|
pt = el.get("PostTypeId")
|
|
if pt == "1":
|
|
qb.append((key, int(el.get("Id")), el.get("Title", ""),
|
|
strip_html(el.get("Body", "")),
|
|
el.get("Tags", ""), int(el.get("Score", 0)),
|
|
int(el.get("AcceptedAnswerId", 0) or 0),
|
|
int(el.get("ViewCount", 0) or 0),
|
|
1 if el.get("ClosedDate") else 0,
|
|
int(el.get("OwnerUserId", 0) or 0)))
|
|
nq += 1
|
|
elif pt == "2":
|
|
ab.append((key, int(el.get("Id")),
|
|
int(el.get("ParentId", 0) or 0),
|
|
strip_html(el.get("Body", "")),
|
|
int(el.get("Score", 0)),
|
|
int(el.get("OwnerUserId", 0) or 0)))
|
|
na += 1
|
|
el.clear()
|
|
if len(qb) >= 2000:
|
|
cn.executemany("INSERT OR REPLACE INTO questions VALUES(?,?,?,?,?,?,?,?,?,?)", qb)
|
|
qb = []
|
|
if len(ab) >= 2000:
|
|
cn.executemany("INSERT OR REPLACE INTO answers VALUES(?,?,?,?,?,?)", ab)
|
|
ab = []
|
|
if qb:
|
|
cn.executemany("INSERT OR REPLACE INTO questions VALUES(?,?,?,?,?,?,?,?,?,?)", qb)
|
|
if ab:
|
|
cn.executemany("INSERT OR REPLACE INTO answers VALUES(?,?,?,?,?,?)", ab)
|
|
|
|
upath = os.path.join(xdir, "Users.xml")
|
|
if os.path.exists(upath):
|
|
ub = []
|
|
for _ev, el in ET.iterparse(upath):
|
|
if el.tag == "row":
|
|
ub.append((key, int(el.get("Id")),
|
|
el.get("DisplayName", "")))
|
|
nu += 1
|
|
el.clear()
|
|
if len(ub) >= 5000:
|
|
cn.executemany("INSERT OR REPLACE INTO users VALUES(?,?,?)", ub)
|
|
ub = []
|
|
if ub:
|
|
cn.executemany("INSERT OR REPLACE INTO users VALUES(?,?,?)", ub)
|
|
|
|
cn.execute("INSERT OR REPLACE INTO ingest_log VALUES(?,?,?,?,?)",
|
|
(key, nq, na, nu, datetime.now().isoformat()))
|
|
cn.commit()
|
|
print(f"{key}: {nq} questions, {na} answers, {nu} users")
|
|
progress("ingest", site=key, questions=nq, answers=na)
|
|
cn.close()
|
|
progress("ingest-done")
|
|
|
|
|
|
def cmd_filter():
|
|
cn = db()
|
|
cn.executescript("""
|
|
DROP TABLE IF EXISTS qa_pairs;
|
|
CREATE TABLE qa_pairs(
|
|
site TEXT, q_id INTEGER, title TEXT, q_body TEXT, tags TEXT,
|
|
q_score INTEGER, view_count INTEGER,
|
|
a_id INTEGER, a_body TEXT, a_score INTEGER, a_accepted INTEGER,
|
|
q_owner INTEGER, a_owner INTEGER,
|
|
PRIMARY KEY(site, q_id, a_id));
|
|
""")
|
|
stats = {}
|
|
for key in SITES:
|
|
before = cn.execute(
|
|
"SELECT COUNT(*) FROM questions WHERE site=?", (key,)).fetchone()[0]
|
|
# accepted + top-scored answer per qualifying question
|
|
cn.execute("""
|
|
INSERT OR REPLACE INTO qa_pairs
|
|
SELECT q.site, q.id, q.title, q.body, q.tags, q.score, q.view_count,
|
|
a.id, a.body, a.score,
|
|
CASE WHEN a.id = q.accepted_answer_id THEN 1 ELSE 0 END,
|
|
q.owner_id, a.owner_id
|
|
FROM questions q
|
|
JOIN answers a ON a.site = q.site AND a.parent_id = q.id
|
|
WHERE q.site = ?
|
|
AND q.score >= ?
|
|
AND q.closed = 0
|
|
AND LENGTH(a.body) >= ?
|
|
AND (a.id = q.accepted_answer_id OR
|
|
a.id = (SELECT a2.id FROM answers a2
|
|
WHERE a2.site = q.site AND a2.parent_id = q.id
|
|
ORDER BY a2.score DESC, a2.id LIMIT 1))
|
|
AND EXISTS (SELECT 1 FROM answers ax
|
|
WHERE ax.site = q.site AND ax.parent_id = q.id
|
|
AND (ax.score >= ? OR ax.id = q.accepted_answer_id))
|
|
""", (key, MIN_Q_SCORE, MIN_ANSWER_CHARS, MIN_A_SCORE))
|
|
after_q = cn.execute(
|
|
"SELECT COUNT(DISTINCT q_id) FROM qa_pairs WHERE site=?",
|
|
(key,)).fetchone()[0]
|
|
after_p = cn.execute(
|
|
"SELECT COUNT(*) FROM qa_pairs WHERE site=?", (key,)).fetchone()[0]
|
|
stats[key] = {"questions_raw": before, "questions_kept": after_q,
|
|
"pairs_kept": after_p}
|
|
print(f"{key}: {before} raw questions -> {after_q} kept "
|
|
f"({after_p} Q&A pairs)")
|
|
cn.commit()
|
|
cn.close()
|
|
json.dump(stats, open(os.path.join(K, "data", "filter-stats.json"), "w",
|
|
encoding="utf-8"), indent=1)
|
|
progress("filter-done", **{k: v["questions_kept"] for k, v in stats.items()})
|
|
|
|
|
|
def cmd_tags():
|
|
cn = db()
|
|
out = os.path.join(K, "data", "tags_freq.csv")
|
|
rows = []
|
|
for key in SITES:
|
|
freq = {}
|
|
for (tags,) in cn.execute(
|
|
"SELECT tags FROM qa_pairs WHERE site=?", (key,)):
|
|
for t in re.findall(r"[<|]([^><|]+)[>|]", tags or ""):
|
|
freq[t] = freq.get(t, 0) + 1
|
|
for t, n in sorted(freq.items(), key=lambda kv: -kv[1]):
|
|
if n >= TAG_MIN_FREQ:
|
|
rows.append((key, t, n))
|
|
with open(out, "w", encoding="utf-8", newline="\n") as f:
|
|
f.write("site,tag,count\n")
|
|
for r in rows:
|
|
f.write(f"{r[0]},{r[1]},{r[2]}\n")
|
|
print(f"{len(rows)} tags with freq >= {TAG_MIN_FREQ} -> {out}")
|
|
per_site = {}
|
|
for s, _t, _n in rows:
|
|
per_site[s] = per_site.get(s, 0) + 1
|
|
print(per_site)
|
|
progress("tags-done", tags_total=len(rows), **per_site)
|
|
cn.close()
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("cmd", choices=["ingest", "filter", "tags",
|
|
"compile", "verify"])
|
|
args = ap.parse_args()
|
|
if args.cmd == "ingest":
|
|
cmd_ingest()
|
|
elif args.cmd == "filter":
|
|
cmd_filter()
|
|
elif args.cmd == "tags":
|
|
cmd_tags()
|
|
elif args.cmd == "compile":
|
|
from se_compile import cmd_compile
|
|
cmd_compile()
|
|
elif args.cmd == "verify":
|
|
from se_verify import cmd_verify
|
|
cmd_verify()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|