Parallel Python 教程：结构化网页研究、任务恢复与引用核验

"""Python 3.11+; no third party packages. Live create is an explicit paid action."""
import argparse
import http.client
import json
import os
import re
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal
from urllib.parse import urlsplit

PRICES = {"core": Decimal("0.025"), "pro": Decimal("0.10"), "ultra": Decimal("0.30")}
SCHEMA = {
    "type": "object", "additionalProperties": False,
    "required": ["company_name", "pricing_page", "features", "unknowns"],
    "properties": {
        "company_name": {"type": "string"},
        "pricing_page": {"type": ["string", "null"]},
        "features": {"type": "array", "maxItems": 5, "items": {"type": "string"}},
        "unknowns": {"type": "array", "items": {"type": "string"}},
    },
}

def connect(path):
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    db.execute("""CREATE TABLE IF NOT EXISTS jobs (
        job_id TEXT PRIMARY KEY, state TEXT NOT NULL, request_json TEXT NOT NULL,
        run_id TEXT, create_reply TEXT, result_json TEXT)""")
    return db

def request_json(method, path, body=None):
    # Fixed provider host; citation URLs are never requested here.
    connection = http.client.HTTPSConnection("api.parallel.ai", timeout=40)
    headers = {"x-api-key": os.environ["PARALLEL_API_KEY"], "Content-Type": "application/json"}
    encoded = None if body is None else json.dumps(body).encode("utf-8")
    try:
        connection.request(method, path, body=encoded, headers=headers)
        response = connection.getresponse()
        if response.status not in (200, 202):
            raise RuntimeError(f"provider HTTP {response.status}; do not resubmit automatically")
        raw = response.read(5_000_001)
        if len(raw) > 5_000_000:
            raise ValueError("response exceeds local size limit")
        value = json.loads(raw)
        if not isinstance(value, dict):
            raise ValueError("expected response object")
        return value
    finally:
        connection.close()

def payload(company, domain, as_of, processor="core"):
    return {
        "input": {"company": company, "official_domain": domain, "as_of": as_of,
                  "objective": "Only research this company on its official site. Return up to five publicly documented product features and its pricing page URL. Use null or unknowns when unverifiable. Do not interpret inaccessible pages as proof of absence. Do not invent publication dates."},
        "processor": processor,
        "source_policy": {"include_domains": [domain]},
        "task_spec": {"output_schema": {"type": "json", "json_schema": SCHEMA}},
    }

def submit(db, job_id, request, limit=Decimal("0.05"), api=request_json):
    cost = PRICES.get(request.get("processor"))
    if cost is None or cost > limit:
        raise ValueError("processor exceeds local per-run budget or is not approved")
    # Commit intent before POST. Duplicate job_id fails before any network call.
    with db:
        db.execute("INSERT INTO jobs(job_id,state,request_json) VALUES(?,?,?)",
                   (job_id, "creating", json.dumps(request, ensure_ascii=False)))
    try:
        reply = api("POST", "/v1/tasks/runs", request)
        with db:
            db.execute("UPDATE jobs SET create_reply=? WHERE job_id=?", (json.dumps(reply), job_id))
        run_id = reply.get("run_id")
        if not isinstance(run_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", run_id):
            raise ValueError("missing or malformed run_id; reconcile create_reply")
        with db:
            db.execute("UPDATE jobs SET state='submitted',run_id=? WHERE job_id=?", (run_id, job_id))
        return run_id
    except Exception:
        with db:
            db.execute("UPDATE jobs SET state='unknown' WHERE job_id=?", (job_id,))
        raise

def collect(db, job_id, api=request_json):
    row = db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
    if row is None or not row["run_id"]:
        raise ValueError("no known run_id; reconcile without creating a replacement")
    result = api("GET", f"/v1/tasks/runs/{row['run_id']}/result?timeout=30")
    # Retain raw result even if subsequent validation fails.
    with db:
        db.execute("UPDATE jobs SET result_json=? WHERE job_id=?", (json.dumps(result), job_id))
    run = result.get("run") or {}
    if run.get("run_id") != row["run_id"] or run.get("status") != "completed":
        raise ValueError("result identity or completion state mismatch")
    with db:
        db.execute("UPDATE jobs SET state='completed' WHERE job_id=?", (job_id,))
    return result

def source_url(value, domain):
    if not isinstance(value, str):
        return False
    try:
        u = urlsplit(value)
        host = (u.hostname or "").lower()
        return (u.scheme == "https" and not u.username and not u.password
                and u.port in (None, 443) and (host == domain or host.endswith("." + domain)))
    except ValueError:
        return False

def evidence_gate(result, expected_company, domain):
    """Preflight only. Never declares a factual claim verified or publishable."""
    issues = []
    output = result.get("output") or {}
    content = output.get("content")
    if output.get("type") != "json" or not isinstance(content, dict):
        return {"state": "blocked", "issues": ["expected JSON output"]}
    fields = set(SCHEMA["required"])
    if set(content) != fields:
        issues.append("schema fields mismatch")
    if content.get("company_name") != expected_company:
        issues.append("entity name mismatch")
    features, unknowns = content.get("features"), content.get("unknowns")
    if not isinstance(features, list) or len(features) > 5 or any(not isinstance(v, str) or not v.strip() for v in features):
        issues.append("invalid features")
        features = []
    if not isinstance(unknowns, list) or any(not isinstance(v, str) or not v.strip() for v in unknowns):
        issues.append("invalid unknowns")
    price_url = content.get("pricing_page")
    if price_url is not None and not source_url(price_url, domain):
        issues.append("pricing URL outside expected official domain")
    basis = output.get("basis")
    mapped = {}
    if not isinstance(basis, list):
        issues.append("missing basis")
        basis = []
    for entry in basis:
        if not isinstance(entry, dict) or not isinstance(entry.get("field"), str):
            issues.append("malformed basis entry")
            continue
        name = entry["field"]
        if name in mapped:
            issues.append("duplicate basis field: " + name)
        mapped[name] = entry
    required = ["company_name"] + (["pricing_page"] if price_url is not None else [])
    required += [f"features.{i}" for i in range(len(features))]
    for field in required:
        entry = mapped.get(field, {})
        citations = entry.get("citations")
        if not isinstance(citations, list) or not citations:
            issues.append("missing citations: " + field)
            continue
        for cite in citations:
            excerpts = cite.get("excerpts") if isinstance(cite, dict) else None
            if (not isinstance(cite, dict) or not source_url(cite.get("url"), domain)
                    or not isinstance(excerpts, list) or not excerpts
                    or any(not isinstance(t, str) or not t.strip() for t in excerpts)):
                issues.append("unusable evidence: " + field)
        if entry.get("confidence") != "high":
            issues.append("confidence needs review: " + field)
    return {"state": "blocked" if issues else "pending_manual_review", "issues": issues,
            "observed_at": datetime.now(timezone.utc).isoformat(),
            "note": "No source was fetched; entity, dates, claim support and conflicts still require human review."}

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=["create", "collect"])
    parser.add_argument("job_id")
    parser.add_argument("--db", default="research.sqlite3")
    args = parser.parse_args()
    with connect(args.db) as db:
        if args.action == "create":
            print(submit(db, args.job_id, payload("Parallel Web Systems", "parallel.ai", "2026-10-16")))
        else:
            result = collect(db, args.job_id)
            print(json.dumps(evidence_gate(result, "Parallel Web Systems", "parallel.ai"), ensure_ascii=False, indent=2))

python3 research.py create parallel-public-profile-001

python3 research.py collect parallel-public-profile-001

{
  "field": "features.0",
  "claim": "待审核的功能描述",
  "source_url": "https://目标官网/具体文档路径",
  "source_published_at": null,
  "observed_at": "由应用记录本次查看时间",
  "supports_claim": null,
  "entity_matches": null,
  "date_scope_matches": null,
  "reviewer": null,
  "decision": "pending"
}

python3 -m unittest discover -s aimazing/assets/parallel -p 'test_research.py' -v