Files
skillfactor-pipeline/pipeline/gen_stats.py
skillfactor-pipeline cb3a876790 feat(knowledge): Stack Exchange as 6th provenance source across the site
stackx category in provenance pies (color #00acc1), landing donut segment +
legend explaining the contribution (practitioner Q&A from six SE communities,
per-entry CC-BY-SA attribution), 6-source chip/formula/pipeline texts,
stats.json stackexchange block, package integration via practitioner-qa.md,
repo-wide ATTRIBUTION.md, verify suite (XML spot checks, mapping consistency,
attribution links, homepage consistency).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PDKeXvpT6tENSvyQGLV1Uq
2026-07-09 22:46:24 +02:00

106 lines
3.4 KiB
Python

"""Generate stats.json — the single source of truth for every displayed count
(landing page, marketplace README, reports). Phase 2c: no number is ever
hard-coded again; whatever the UI shows comes from this file.
Writes:
data/stats.json (versioned artifact)
<gitea>/custom/public/assets/stats.json (served to the landing page)
Run after enrichment/aggregation changes: python pipeline/gen_stats.py
"""
import json
import os
import sys
from datetime import datetime
sys.path.insert(0, os.path.dirname(__file__))
BASE = os.path.join(os.path.dirname(__file__), "..")
SKILLS = os.path.join(BASE, "skills")
OUT = os.path.join(BASE, "data", "stats.json")
ASSETS_OUT = (r"C:\Program Files (x86)\TempoBill\Zeiterfassung.Cloud"
r"\Tools\gitea\custom\public\assets\stats.json")
FLAGSHIP = "artificial-intelligence-engineer"
def main():
total = white = enriched = mapped_links = 0
flagship = {}
for slug in os.listdir(SKILLS):
mp = os.path.join(SKILLS, slug, "manifest.json")
if not os.path.isfile(mp):
continue
try:
m = json.load(open(mp, encoding="utf-8"))
except ValueError:
continue
total += 1
if m.get("collar") == "white":
white += 1
enr = m.get("enrichment_ai_skills") or {}
n = enr.get("total_skills", 0)
if n:
enriched += 1
mapped_links += n
if slug == FLAGSHIP:
flagship = {
"slug": slug,
"ai_skills_total": n,
"tiers": enr.get("tiers", {}),
"provenance_items": (m.get("provenance") or {}).get("items", {}),
}
# occupation-to-skill relations + external catalog size
relations = 0
try:
from db import connect
cn = connect()
cur = cn.cursor()
cur.execute("SELECT COUNT(*) FROM esco_occ_skill")
relations = cur.fetchone()[0]
cn.close()
except Exception:
pass
ext_sources = ext_skills = 0
try:
import p5_enrich_ai_skills as p5
catalog = p5.load_catalog()
ext_sources = len(p5.ALL_SOURCES)
ext_skills = sum(len(v) for v in catalog.values())
except Exception:
pass
# Stack-Exchange-Wissensschicht (knowledge/): kompilierte Q&A-Einträge
se = {}
se_stats_p = os.path.join(BASE, "knowledge", "data", "compile-stats.json")
if os.path.exists(se_stats_p):
try:
cs = json.load(open(se_stats_p, encoding="utf-8"))
se = {"professions": len(cs),
"qa_entries": sum(v.get("entries", 0) for v in cs.values())}
except ValueError:
pass
stats = {
"generated_at": datetime.now().isoformat(timespec="seconds"),
"stackexchange": se,
"total_occupations": total,
"white_collar": white,
"occupation_skill_relations": relations,
"external_sources": ext_sources,
"external_skills": ext_skills,
"enriched_packages": enriched,
"mapped_skill_links": mapped_links,
"flagship": flagship,
}
json.dump(stats, open(OUT, "w", encoding="utf-8"), indent=2)
try:
json.dump(stats, open(ASSETS_OUT, "w", encoding="utf-8"), indent=2)
except OSError as exc:
print(f"WARN: assets copy failed: {exc}")
print(json.dumps(stats, indent=1)[:600])
if __name__ == "__main__":
main()