feat: add container manager API, islop router, and container runtime files with vim/bot/d stealth clients

devplacepy/routers/tools/isslop.py

@@ -0,0 +1,506 @@
+# retoor <retoor@molodetz.nl>
+
+import logging
+import re
+from urllib.parse import quote
+from typing import Annotated
+
+import uuid_utils
+from fastapi import Depends, APIRouter, Request
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
+
+from devplacepy.config import ISSLOP_MEDIA_DIR
+from devplacepy.constants import DEVII_GUEST_COOKIE
+from devplacepy.dependencies import json_or_form
+from devplacepy.models import IsslopRunForm
+from devplacepy.responses import respond
+from devplacepy.schemas import IsslopAnalysisOut, IsslopListOut, IsslopReportOut, IsslopSourceOut
+from devplacepy.seo import base_seo_context, site_url, web_application_schema, website_schema
+from devplacepy.services.jobs import queue
+from devplacepy.services.jobs.isslop import store
+from devplacepy.services.jobs.isslop.badge import badge_html, badge_markdown, render_badge
+from devplacepy.services.jobs.isslop.service import topic_for
+from devplacepy.templating import templates
+from devplacepy.utils import get_current_user, not_found, track_action
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+ACTIVE_STATES = (queue.PENDING, queue.RUNNING)
+GUEST_COOKIE_MAX_AGE = 31536000
+EVENT_LIMIT_MAX = 5000
+MEDIA_NAME_PATTERN = re.compile(r"^[a-f0-9]{16}\.webp$")
+SOURCE_NAME_PATTERN = re.compile(r"^s[a-f0-9]{16}\.txt$")
+SOURCE_LINE_CONTEXT = 400000
+
+
+def _owner(request: Request) -> tuple[str, str] | None:
+    user = get_current_user(request)
+    if user:
+        return "user", user["uid"]
+    guest = request.cookies.get(DEVII_GUEST_COOKIE)
+    if guest:
+        return "guest", guest
+    return None
+
+
+def _ensure_owner(request: Request) -> tuple[str, str, str]:
+    owner = _owner(request)
+    if owner:
+        return owner[0], owner[1], ""
+    minted = uuid_utils.uuid7().hex
+    return "guest", minted, minted
+
+
+def _set_guest_cookie(response, minted: str) -> None:
+    if minted:
+        response.set_cookie(
+            DEVII_GUEST_COOKIE,
+            minted,
+            httponly=True,
+            samesite="lax",
+            max_age=GUEST_COOKIE_MAX_AGE,
+        )
+
+
+def _sync_guest_history(request: Request) -> None:
+    user = get_current_user(request)
+    guest = request.cookies.get(DEVII_GUEST_COOKIE)
+    if user and guest:
+        store.claim_guest_analyses(guest, user["uid"])
+
+
+def _analysis_payload(row: dict) -> dict:
+    uid = row.get("uid", "")
+    return {
+        "uid": uid,
+        "status": row.get("status", ""),
+        "source_url": row.get("source_url", ""),
+        "source_kind": row.get("source_kind", "") or "unknown",
+        "grade": row.get("grade"),
+        "slop_score": row.get("slop_score"),
+        "origin_score": row.get("origin_score"),
+        "quality_deficit_score": row.get("quality_deficit_score"),
+        "human_percent": row.get("human_percent"),
+        "ai_percent": row.get("ai_percent"),
+        "category": row.get("category"),
+        "confidence": row.get("confidence"),
+        "files_total": int(row.get("files_total") or 0),
+        "files_analyzed": int(row.get("files_analyzed") or 0),
+        "detected_builder": row.get("detected_builder") or None,
+        "dom_slop_score": row.get("dom_slop_score"),
+        "error": row.get("error_message_text") or None,
+        "report_url": f"/tools/isslop/{uid}/report",
+        "badge_url": f"/tools/isslop/{uid}/badge.svg",
+        "events_url": f"/tools/isslop/{uid}/events",
+        "topic": topic_for(uid),
+        "created_at": row.get("created_at"),
+        "finished_at": row.get("finished_at"),
+    }
+
+
+SEVERITY_RANK = {"strong": 0, "medium": 1, "weak": 2}
+SIGNAL_LINES_CAP = 8
+
+
+def _signal_groups(signals: list) -> list:
+    groups: dict[str, dict] = {}
+    for signal in signals:
+        if not isinstance(signal, dict):
+            continue
+        code = str(signal.get("code", ""))
+        entry = groups.setdefault(
+            code,
+            {
+                "code": code,
+                "title": str(signal.get("title", "")),
+                "severity": str(signal.get("severity", "weak")),
+                "count": 0,
+                "lines": [],
+            },
+        )
+        entry["count"] += 1
+        line = signal.get("line")
+        if isinstance(line, int) and len(entry["lines"]) < SIGNAL_LINES_CAP:
+            entry["lines"].append(line)
+    return sorted(
+        groups.values(),
+        key=lambda group: (SEVERITY_RANK.get(group["severity"], 3), -group["count"], group["code"]),
+    )
+
+
+def _source_url(uid: str, path: str, line: int = 0) -> str:
+    url = f"/tools/isslop/{uid}/source?path={quote(path, safe='')}"
+    if line > 0:
+        url += f"&line={line}#L{line}"
+    return url
+
+
+CRITERIA_URL = "/docs/isslop-checks.html"
+
+
+def _linkify_sources(markdown: str, uid: str, paths: set[str], signal_codes: set[str]) -> str:
+    for path in sorted(paths, key=len, reverse=True):
+        escaped = re.escape(path)
+        markdown = re.sub(
+            rf"`{escaped}:(\d+)`",
+            lambda match, p=path: f"[`{p}:{match.group(1)}`]({_source_url(uid, p, int(match.group(1)))})",
+            markdown,
+        )
+        markdown = markdown.replace(f"`{path}`", f"[`{path}`]({_source_url(uid, path)})")
+        markdown = re.sub(
+            rf"(?<![\w/`\(\[]){escaped}(?::(\d+))?(?![\w/`])",
+            lambda match, p=path: (
+                f"[{p}:{match.group(1)}]({_source_url(uid, p, int(match.group(1)))})"
+                if match.group(1)
+                else f"[{p}]({_source_url(uid, p)})"
+            ),
+            markdown,
+        )
+    for code in sorted(signal_codes, key=len, reverse=True):
+        escaped = re.escape(code)
+        markdown = markdown.replace(f"`{code}`", f"[`{code}`]({CRITERIA_URL})")
+        markdown = re.sub(
+            rf"(?<![\w`\[]){escaped}(?![\w`])",
+            f"[{code}]({CRITERIA_URL})",
+            markdown,
+        )
+    return markdown
+
+
+def _badge_info(request: Request, uid: str) -> dict:
+    base = site_url(request).rstrip("/")
+    badge_url = f"{base}/tools/isslop/{uid}/badge.svg"
+    report_url = f"{base}/tools/isslop/{uid}/report"
+    return {
+        "badge_url": badge_url,
+        "report_url": report_url,
+        "markdown": badge_markdown(badge_url, report_url),
+        "html": badge_html(badge_url, report_url),
+    }
+
+
+@router.get("", response_class=HTMLResponse)
+async def isslop_page(request: Request):
+    user = get_current_user(request)
+    _sync_guest_history(request)
+    base = site_url(request)
+    description = (
+        "Measure how a codebase or website was made: untouched AI defaults, AI steered by a "
+        "knowing hand, or work no model would ever produce. Transparent, reproducible analysis "
+        "with a shareable authenticity badge."
+    )
+    seo_ctx = base_seo_context(
+        request,
+        title="AI Usage Analyzer",
+        description=description,
+        breadcrumbs=[
+            {"name": "Home", "url": "/feed"},
+            {"name": "Tools", "url": "/tools"},
+            {"name": "AI Usage Analyzer", "url": "/tools/isslop"},
+        ],
+        schemas=[
+            website_schema(base),
+            web_application_schema("AI Usage Analyzer", description, "/tools/isslop", base),
+        ],
+    )
+    minted = "" if _owner(request) else uuid_utils.uuid7().hex
+    response = templates.TemplateResponse(
+        request,
+        "tools/isslop.html",
+        {**seo_ctx, "request": request, "user": user},
+    )
+    _set_guest_cookie(response, minted)
+    return response
+
+
+@router.post("/run")
+async def isslop_run(request: Request, data: Annotated[IsslopRunForm, Depends(json_or_form(IsslopRunForm))]):
+    from devplacepy.services.audit import record as audit
+
+    owner_kind, owner_id, minted = _ensure_owner(request)
+    active = [
+        job
+        for job in queue.list_jobs(kind="isslop", owner=(owner_kind, owner_id))
+        if job.get("status") in ACTIVE_STATES
+    ]
+    if active:
+        audit.record(
+            request,
+            "isslop.run.request",
+            result="denied",
+            summary=f"AI usage analysis denied for {data.url}: analysis already running",
+            metadata={"target": data.url, "reason": "active_job", "uid": active[0]["uid"]},
+            links=[audit.job(active[0]["uid"])],
+        )
+        return JSONResponse(
+            {
+                "error": {
+                    "status": 429,
+                    "message": "You already have an analysis running. Wait for it to finish.",
+                    "uid": active[0]["uid"],
+                }
+            },
+            status_code=429,
+        )
+    uid = queue.enqueue(
+        "isslop",
+        {"url": data.url},
+        owner_kind,
+        owner_id,
+        f"AI usage: {data.url}"[:64],
+    )
+    store.create_analysis(uid, data.url, owner_kind, owner_id)
+    audit.record(
+        request,
+        "isslop.run.request",
+        summary=f"requested AI usage analysis of {data.url}",
+        metadata={"target": data.url},
+        links=[audit.job(uid)],
+    )
+    if owner_kind == "user":
+        track_action(owner_id, "isslop")
+    response = JSONResponse(
+        {
+            "uid": uid,
+            "status_url": f"/tools/isslop/{uid}",
+            "events_url": f"/tools/isslop/{uid}/events",
+            "report_url": f"/tools/isslop/{uid}/report",
+            "topic": topic_for(uid),
+        }
+    )
+    _set_guest_cookie(response, minted)
+    return response
+
+
+@router.get("/list")
+async def isslop_list(request: Request, limit: int = store.LIST_LIMIT_DEFAULT):
+    _sync_guest_history(request)
+    owner = _owner(request)
+    rows = []
+    if owner:
+        rows = store.list_analyses(owner[0], owner[1], min(max(1, limit), 200))
+    return JSONResponse(
+        IsslopListOut.model_validate(
+            {"analyses": [_analysis_payload(row) for row in rows]}
+        ).model_dump(mode="json")
+    )
+
+
+@router.get("/{uid}")
+async def isslop_status(request: Request, uid: str):
+    row = store.get_analysis(uid)
+    if not row:
+        raise not_found("Analysis not found")
+    return JSONResponse(
+        IsslopAnalysisOut.model_validate(_analysis_payload(row)).model_dump(mode="json")
+    )
+
+
+@router.get("/{uid}/events")
+async def isslop_events(request: Request, uid: str, after: int = 0, limit: int = store.EVENT_LIMIT_DEFAULT):
+    row = store.get_analysis(uid)
+    if not row:
+        raise not_found("Analysis not found")
+    rows = store.events_for(uid, max(0, after), min(max(1, limit), EVENT_LIMIT_MAX))
+    events = [
+        {
+            "seq": event["seq"],
+            "kind": event["kind"],
+            "message": event["message"],
+            "data": store.decode_json(event.get("payload"), {}),
+            "created_at": event["created_at"],
+        }
+        for event in rows
+    ]
+    return JSONResponse({"uid": uid, "status": row.get("status", ""), "events": events})
+
+
+@router.get("/{uid}/report")
+async def isslop_report(request: Request, uid: str):
+    row = store.get_analysis(uid)
+    if not row:
+        raise not_found("Analysis not found")
+    report = store.get_report(uid)
+    files = []
+    linkable_paths: set[str] = set()
+    signal_codes: set[str] = set()
+    for item in store.file_results_for(uid):
+        signals = store.decode_json(item.get("signals"), [])
+        for signal in signals:
+            if isinstance(signal, dict) and signal.get("code"):
+                signal_codes.add(str(signal["code"]))
+        source_name = str(item.get("source") or "")
+        has_source = bool(source_name and SOURCE_NAME_PATTERN.match(source_name))
+        path = item.get("path", "")
+        if has_source:
+            linkable_paths.add(path)
+        files.append(
+            {
+                "path": path,
+                "language": item.get("language", "unknown"),
+                "lines": int(item.get("lines") or 0),
+                "origin_score": float(item.get("origin_score") or 0.0),
+                "quality_deficit_score": float(item.get("quality_deficit_score") or 0.0),
+                "category": item.get("category", "uncertain"),
+                "signals": signals,
+                "signal_groups": _signal_groups(signals),
+                "source_url": _source_url(uid, path) if has_source else None,
+            }
+        )
+    images = [
+        {
+            "path": item.get("path", ""),
+            "ai_probability": float(item.get("ai_probability") or 0.0),
+            "grade": item.get("grade", "n/a"),
+            "verdict": item.get("verdict", "uncertain"),
+            "image_kind": item.get("image_kind", "image"),
+            "tells": store.decode_json(item.get("tells"), []),
+            "description": item.get("description", ""),
+            "thumb_url": (
+                f"/tools/isslop/{uid}/media/{item['thumb']}"
+                if item.get("thumb") and MEDIA_NAME_PATTERN.match(str(item["thumb"]))
+                else None
+            ),
+        }
+        for item in store.image_results_for(uid)
+    ]
+    dom_pages = [
+        {
+            "url": item.get("url", ""),
+            "detected_builder": item.get("detected_builder") or None,
+            "signal_count": int(item.get("signal_count") or 0),
+            "signals": store.decode_json(item.get("signals"), []),
+            "screenshot_url": (
+                f"/tools/isslop/{uid}/media/{item['screenshot']}"
+                if item.get("screenshot") and MEDIA_NAME_PATTERN.match(str(item["screenshot"]))
+                else None
+            ),
+        }
+        for item in store.dom_results_for(uid)
+    ]
+    seo_ctx = base_seo_context(
+        request,
+        title="AI usage analysis report",
+        description=f"Authenticity analysis of {row.get('source_url', '')}",
+        robots="noindex,nofollow",
+        breadcrumbs=[
+            {"name": "Home", "url": "/feed"},
+            {"name": "Tools", "url": "/tools"},
+            {"name": "AI Usage Analyzer", "url": "/tools/isslop"},
+            {"name": "Report", "url": f"/tools/isslop/{uid}/report"},
+        ],
+    )
+    context = {
+        **seo_ctx,
+        **_analysis_payload(row),
+        "content_hash": row.get("content_hash"),
+        "markdown": _linkify_sources(report.get("markdown", ""), uid, linkable_paths, signal_codes) if report else "",
+        "generator_model": report.get("model_used", "") if report else "",
+        "generated_at": report.get("generated_at") if report else None,
+        "badge": _badge_info(request, uid),
+        "files": files,
+        "images": images,
+        "dom_pages": dom_pages,
+        "request": request,
+        "user": get_current_user(request),
+    }
+    return respond(request, "tools/isslop_report.html", context, model=IsslopReportOut)
+
+
+@router.get("/{uid}/report.md")
+async def isslop_report_markdown(request: Request, uid: str):
+    row = store.get_analysis(uid)
+    if not row:
+        raise not_found("Analysis not found")
+    report = store.get_report(uid)
+    if not report:
+        raise not_found("Report not yet generated for this analysis")
+    return Response(
+        content=report["markdown"],
+        media_type="text/markdown; charset=utf-8",
+        headers={"content-disposition": f'attachment; filename="ai-usage-report-{uid}.md"'},
+    )
+
+
+@router.get("/{uid}/source")
+async def isslop_source(request: Request, uid: str, path: str, line: int = 0):
+    row = store.get_analysis(uid)
+    if not row:
+        raise not_found("Analysis not found")
+    result = store.file_result_for(uid, path)
+    source_name = str(result.get("source") or "") if result else ""
+    if not result or not SOURCE_NAME_PATTERN.match(source_name):
+        raise not_found("Source not available for this file")
+    source_path = (store.media_dir_for(uid) / source_name).resolve()
+    if not source_path.is_relative_to(ISSLOP_MEDIA_DIR.resolve()) or not source_path.is_file():
+        raise not_found("Source not available for this file")
+    text = source_path.read_text(encoding="utf-8", errors="replace")
+    signals = store.decode_json(result.get("signals"), [])
+    marked: dict[int, list] = {}
+    for signal in signals:
+        if isinstance(signal, dict) and isinstance(signal.get("line"), int) and signal["line"] > 0:
+            marked.setdefault(signal["line"], []).append(signal)
+    source_lines = text.split("\n")
+    seo_ctx = base_seo_context(
+        request,
+        title=f"Source: {path}",
+        description=f"Annotated source of {path} from the AI usage analysis",
+        robots="noindex,nofollow",
+        breadcrumbs=[
+            {"name": "Home", "url": "/feed"},
+            {"name": "Tools", "url": "/tools"},
+            {"name": "AI Usage Analyzer", "url": "/tools/isslop"},
+            {"name": "Report", "url": f"/tools/isslop/{uid}/report"},
+            {"name": "Source", "url": _source_url(uid, path)},
+        ],
+    )
+    context = {
+        **seo_ctx,
+        "uid": uid,
+        "path": path,
+        "language": result.get("language", "unknown"),
+        "category": result.get("category", "uncertain"),
+        "origin_score": float(result.get("origin_score") or 0.0),
+        "quality_deficit_score": float(result.get("quality_deficit_score") or 0.0),
+        "source": text,
+        "truncated": len(text) >= SOURCE_LINE_CONTEXT,
+        "signals": signals,
+        "source_lines": source_lines,
+        "marked_lines": marked,
+        "focus_line": max(0, line),
+        "report_url": f"/tools/isslop/{uid}/report",
+        "request": request,
+        "user": get_current_user(request),
+    }
+    return respond(request, "tools/isslop_source.html", context, model=IsslopSourceOut)
+
+
+@router.get("/{uid}/media/{name}")
+async def isslop_media(request: Request, uid: str, name: str):
+    row = store.get_analysis(uid)
+    if not row or not MEDIA_NAME_PATTERN.match(name):
+        raise not_found("Image not available")
+    root = ISSLOP_MEDIA_DIR.resolve()
+    path = (store.media_dir_for(uid) / name).resolve()
+    if not path.is_relative_to(root) or not path.is_file():
+        raise not_found("Image not available")
+    return FileResponse(
+        path,
+        media_type="image/webp",
+        headers={"cache-control": "public, max-age=86400"},
+    )
+
+
+@router.get("/{uid}/badge.svg")
+async def isslop_badge(request: Request, uid: str):
+    row = store.get_analysis(uid)
+    if not row:
+        raise not_found("Analysis not found")
+    report_url = _badge_info(request, uid)["report_url"]
+    svg = render_badge(row.get("human_percent"), row.get("grade"), report_url)
+    return Response(
+        content=svg,
+        media_type="image/svg+xml",
+        headers={"cache-control": "no-cache, max-age=300"},
+    )