#!/usr/bin/env python3 """GEO Evidence Kit: explicit citations, source roles and independent API requests. Python 3.10+, standard library only. No URL fetching during extract/report/plan. Only `run --execute` can call the fixed OpenAI Responses endpoint. """ import argparse import copy import datetime as dt import ipaddress import json import os from pathlib import Path import re import tempfile import unicodedata import urllib.error import urllib.parse import urllib.request VERSION = "1.0.0" MAX_BYTES = 16 * 1024 * 1024 ENDPOINT = "https://api.openai.com/v1/responses" PAGE_TYPES = {"brand", "retailer", "forum", "video", "reference", "other", "unclassified"} class EvidenceError(ValueError): """Invalid input, unsupported export or unsafe request plan.""" def now(): return dt.datetime.now(dt.timezone.utc).isoformat() def read_json(path): with open(path, "rb") as stream: raw = stream.read(MAX_BYTES + 1) if len(raw) > MAX_BYTES: raise EvidenceError("JSON input exceeds 16 MiB") try: return json.loads(raw) except (ValueError, UnicodeError) as exc: raise EvidenceError("Input must be valid UTF-8 JSON") from exc def write_json(path, value, *, replace=False): """Atomic, private output; existing files require an explicit internal replace.""" target = Path(path) if not target.parent.is_dir(): raise EvidenceError("Output parent directory does not exist") serialized = json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n" if not replace: fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) os.close(fd) temp = None try: with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=target.parent, prefix=".geo-evidence-", delete=False) as stream: temp = Path(stream.name) stream.write(serialized) os.replace(temp, target) finally: if temp and temp.exists(): temp.unlink() def safe_page_url(value): """Validate without network access. Preserve query; strip fragment for grouping.""" if not isinstance(value, str) or not value or len(value) > 8192: raise EvidenceError("URL must be a nonempty string of at most 8192 characters") if any(c.isspace() or ord(c) < 32 or ord(c) == 127 for c in value) or "\\" in value: raise EvidenceError("URL contains whitespace, control characters or backslashes") if re.search(r"%(?:0[0-9a-f]|1[0-9a-f]|7f)", value, re.I): raise EvidenceError("URL contains encoded control characters") try: parts = urllib.parse.urlsplit(value) if parts.scheme.lower() not in ("http", "https") or not parts.hostname: raise EvidenceError("Only absolute HTTP(S) URLs are supported") if parts.username is not None or parts.password is not None: raise EvidenceError("URL credentials are not allowed") host = parts.hostname.encode("idna").decode("ascii").lower() port = parts.port # validates malformed and out-of-range ports try: address = ipaddress.ip_address(host) except ValueError: if "." not in host or host.endswith((".localhost", ".local", ".internal")): raise EvidenceError("Local hostnames are not supported") if any(not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", p) for p in host.split(".")): raise EvidenceError("Invalid domain name") else: if not address.is_global: raise EvidenceError("Non-public IP addresses are not supported") if address.version == 6: host = "[" + host + "]" scheme = parts.scheme.lower() netloc = host if port in (None, 443 if scheme == "https" else 80) else f"{host}:{port}" return urllib.parse.urlunsplit((scheme, netloc, parts.path or "/", parts.query, "")) except (ValueError, UnicodeError) as exc: if isinstance(exc, EvidenceError): raise raise EvidenceError("Malformed URL") from exc def extract_response(raw, mention_terms=()): """Parse raw Responses JSON or an explicitly human-reviewed manual export.""" if not isinstance(raw, dict): raise EvidenceError("Expected a single response/export object, not a stream or list") if not isinstance(mention_terms, (list, tuple)) or any(not isinstance(t, str) for t in mention_terms): raise EvidenceError("Mention terms must be strings") result = {"schema": "gptfav.evidence.v1", "version": VERSION, "checked_at": now(), "raw_response": copy.deepcopy(raw), "answer_blocks": [], "citations": [], "candidates": [], "mentions": [], "rejected": [], "warnings": [], "network_fetches": 0, "mention_terms": list(mention_terms)} manual = raw.get("schema") == "gptfav.manual-citations.v1" page_types = {} if manual: review = raw.get("review", {}) required = ("reviewed_by", "reviewed_at", "evidence_reference") if not isinstance(review, dict) or review.get("confirmed") is not True or any( not isinstance(review.get(k), str) or not review[k].strip() for k in required): raise EvidenceError("Manual export needs a named reviewer, date, evidence reference and confirmed=true") if not isinstance(raw.get("answer_text"), str) or not isinstance(raw.get("citations"), list): raise EvidenceError("Manual export needs answer_text and a citations array") if raw.get("offset_unit") != "unicode_codepoints": raise EvidenceError("Manual offsets must declare unicode_codepoints") if not isinstance(raw.get("page_types", {}), dict): raise EvidenceError("Manual page_types must map page URLs to human-supplied type labels") for url, label in raw.get("page_types", {}).items(): page = safe_page_url(url) if not isinstance(label, str) or label not in PAGE_TYPES: raise EvidenceError("Unsupported page type label") if page in page_types and page_types[page] != label: raise EvidenceError("Conflicting manual type labels for the same page") page_types[page] = label annotations = [{**a, "type": "url_citation"} for a in raw["citations"] if isinstance(a, dict)] if len(annotations) != len(raw["citations"]): raise EvidenceError("Manual citation entries must be objects") outputs = [{"type": "message", "role": "assistant", "content": [ {"type": "output_text", "text": raw["answer_text"], "annotations": annotations}]}] result.update(source_format="human-reviewed-manual", response_id=raw.get("response_id"), response_status="human-reviewed", model=raw.get("model"), human_review=review, question=raw.get("question")) else: if raw.get("object") != "response" or not isinstance(raw.get("output"), list): raise EvidenceError("Unsupported export: expected Responses object/output or explicit manual schema") outputs = raw["output"] result.update(source_format="openai-responses", response_id=raw.get("id"), response_status=raw.get("status"), model=raw.get("model")) if raw.get("status") != "completed": result["warnings"].append("Response is not completed; this is partial evidence, not a successful trial") def candidate(source, location): if not isinstance(source, dict): result["rejected"].append({"role": "candidate", "location": location, "reason": "Expected source object"}) return try: page = safe_page_url(source.get("url")) except EvidenceError as exc: result["rejected"].append({"role": "candidate", "location": location, "reason": str(exc)}) return result["candidates"].append({"role": "search_candidate", "url": source["url"], "page_url": page, "title": source.get("title"), "location": location}) for oi, output in enumerate(outputs): if not isinstance(output, dict): result["warnings"].append(f"output[{oi}] is not an object") continue if output.get("type") == "web_search_call": action = output.get("action") sources = action.get("sources", []) if isinstance(action, dict) else [] if isinstance(sources, list): for si, source in enumerate(sources): candidate(source, f"output[{oi}].action.sources[{si}]") else: result["warnings"].append(f"output[{oi}].action.sources is not an array") if output.get("type") != "message" or output.get("role") != "assistant": continue contents = output.get("content", []) if not isinstance(contents, list): raise EvidenceError("Assistant content must be an array") for ci, content in enumerate(contents): if not isinstance(content, dict) or content.get("type") != "output_text": continue text = content.get("text") annotations = content.get("annotations", []) if not isinstance(text, str) or not isinstance(annotations, list): raise EvidenceError("output_text requires text and an annotations array") block = f"output[{oi}].content[{ci}]" result["answer_blocks"].append({"location": block, "text": text}) if not manual and any(ord(c) > 0xffff for c in text): result["warnings"].append(f"{block} contains non-BMP characters; API offsets are interpreted as Unicode codepoints, verify against the original answer") cited_pages = set() for ai, annotation in enumerate(annotations): if not isinstance(annotation, dict) or annotation.get("type") != "url_citation": continue location = f"{block}.annotations[{ai}]" try: page = safe_page_url(annotation.get("url")) start, end = annotation.get("start_index"), annotation.get("end_index") if type(start) is not int or type(end) is not int or not (0 <= start < end <= len(text)): raise EvidenceError("Citation offsets must select a nonempty span inside the answer") except EvidenceError as exc: result["rejected"].append({"role": "citation", "location": location, "reason": str(exc), "annotation": annotation}) continue cited_pages.add(page) result["citations"].append({"role": "answer_citation", "url": annotation["url"], "page_url": page, "title": annotation.get("title"), "location": location, "page_type": page_types.get(page, "unclassified"), "page_type_source": "human-review" if page in page_types else "not-classified", "block": block, "start_index": start, "end_index": end, "offset_unit": "unicode_codepoints_interpreted", "excerpt": text[start:end], "answer_text": text, "raw_annotation": copy.deepcopy(annotation)}) for match in re.finditer(r"https?://[^\s<>\]\)\"'，。；]+", text): url = match.group().rstrip(".,;:!?") try: page = safe_page_url(url) except EvidenceError: continue if page not in cited_pages: result["mentions"].append({"role": "unannotated_url_mention", "url": url, "page_url": page, "block": block, "start_index": match.start(), "end_index": match.start() + len(url)}) for term in mention_terms: if isinstance(term, str) and term: for match in re.finditer(re.escape(term), text, re.I): result["mentions"].append({"role": "literal_term_mention", "term": term, "block": block, "start_index": match.start(), "end_index": match.end()}) if manual: candidates = raw.get("candidates", []) if not isinstance(candidates, list): raise EvidenceError("Manual candidates must be an array") for i, source in enumerate(candidates): candidate(source, f"candidates[{i}]") if not result["answer_blocks"]: result["warnings"].append("No assistant output_text blocks: no answer citation evidence") if manual and any(r["role"] == "citation" for r in result["rejected"]): raise EvidenceError("Manual citation contains an invalid URL or body span; correct the export") return result def report_evidence(documents): """Re-parse preserved raw objects, so edited derived counts cannot add evidence.""" pages, candidates, mentions, rejected, warnings = {}, [], [], [], [] records, runs, failures = [], [], [] for source_number, doc in enumerate(documents): if isinstance(doc, dict) and doc.get("schema") == "gptfav.run.v1": if doc.get("mode") != "execute": raise EvidenceError("A dry run is a plan, not response evidence") entries = doc.get("records") planned = doc.get("planned_requests") if not isinstance(entries, list) or not isinstance(planned, list): raise EvidenceError("Run file needs records and planned_requests arrays") completed = 0 for entry in entries: if not isinstance(entry, dict): raise EvidenceError("Run records must be objects") if entry.get("status") == "completed": completed += 1 else: error = entry.get("error", {}) failures.append({"source_document": source_number, "question_id": entry.get("question_id"), "status": entry.get("status"), "error": {k: error[k] for k in ("type", "http_status", "response_status") if isinstance(error, dict) and k in error}, "has_raw_response": "raw_response" in entry}) if "raw_response" in entry: payload = entry.get("payload", {}) records.append({"document": entry["raw_response"], "from_run": True, "source_document": source_number, "question_id": entry.get("question_id"), "question": payload.get("input") if isinstance(payload, dict) else None, "request_model": payload.get("model") if isinstance(payload, dict) else None, "run_started_at": doc.get("started_at"), "run_finished_at": doc.get("finished_at"), "request_started_at": entry.get("started_at"), "request_finished_at": entry.get("finished_at")}) runs.append({"source_document": source_number, "planned": len(planned), "attempted": len(entries), "completed_recorded": completed, "failed_recorded": len(entries) - completed, "unattempted": max(0, len(planned) - len(entries)), "records_without_raw_response": sum("raw_response" not in e for e in entries)}) else: records.append({"document": doc, "from_run": False, "source_document": source_number}) for number, record in enumerate(records): doc = record["document"] raw = doc.get("raw_response") if isinstance(doc, dict) and doc.get("schema") == "gptfav.evidence.v1" else doc terms = doc.get("mention_terms", []) if isinstance(doc, dict) and doc.get("schema") == "gptfav.evidence.v1" else [] try: evidence = extract_response(raw, terms) except EvidenceError as exc: if not record["from_run"]: raise rejected.append({"input_record": number, "source_document": record["source_document"], "question_id": record.get("question_id"), "role": "raw_response", "reason": str(exc)}) continue label = evidence.get("response_id") or f"input-{number + 1}" context = {key: value for key, value in record.items() if key not in ("document", "from_run")} context.update(model=evidence.get("model"), human_review=evidence.get("human_review")) context.setdefault("question", evidence.get("question")) for citation in evidence["citations"]: entry = pages.setdefault(citation["page_url"], {"page_url": citation["page_url"], "page_type": "unclassified", "page_type_labels": [], "citation_occurrences": 0, "input_records": [], "evidence": []}) if citation["page_type"] not in entry["page_type_labels"]: entry["page_type_labels"].append(citation["page_type"]) entry["page_type"] = entry["page_type_labels"][0] if len(entry["page_type_labels"]) == 1 else "unclassified" entry["citation_occurrences"] += 1 if number not in entry["input_records"]: entry["input_records"].append(number) entry["evidence"].append({"input_record": number, "response_id": label, "response_status": evidence["response_status"], "source_format": evidence["source_format"], **context, **citation}) for key, target in (("candidates", candidates), ("mentions", mentions), ("rejected", rejected)): target.extend({"input_record": number, "response_id": label, **context, **entry} for entry in evidence[key]) warnings.extend({"input_record": number, "warning": message} for message in evidence["warnings"]) return {"schema": "gptfav.report.v1", "version": VERSION, "created_at": now(), "source_documents": len(documents), "input_records": len(records), "runs": runs, "execution_failures": failures, "cited_pages": sorted(pages.values(), key=lambda p: p["page_url"]), "search_candidates": candidates, "mentions": mentions, "rejected": rejected, "warnings": warnings, "limitations": ["Only answer url_citation annotations or explicitly reviewed body citations count as citations", "No external pages were fetched; page types are unclassified unless a manual export supplies human labels; citations do not establish that a claim is true", "Input record counts are not independent trial counts; duplicate exports may be present", "Run summaries preserve recorded failures and unattempted questions; no citation success rate is inferred", "API results and manual ChatGPT webpage exports are different evidence sources"]} def folded(value): return "".join(c for c in unicodedata.normalize("NFKC", value).casefold() if c.isalnum()) def make_plan(config): if not isinstance(config, dict): raise EvidenceError("Plan config must be an object") allowed = {"model", "questions", "excluded_brand_terms", "web_search", "max_output_tokens"} if set(config) - allowed: raise EvidenceError("Config contains unsupported fields; conversation/history/instructions are not accepted") model, questions, terms = config.get("model"), config.get("questions"), config.get("excluded_brand_terms") if not isinstance(model, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,119}", model): raise EvidenceError("Supply an explicit model ID; no model is selected automatically") if not isinstance(questions, list) or not 1 <= len(questions) <= 100: raise EvidenceError("Supply 1 to 100 questions") if not isinstance(terms, list) or not terms or any(not isinstance(t, str) or not folded(t) for t in terms): raise EvidenceError("Provide nonempty brand/alias terms for the local leakage check") web = config.get("web_search", False) tokens = config.get("max_output_tokens", 2000) if type(web) is not bool or type(tokens) is not int or not 16 <= tokens <= 10000: raise EvidenceError("web_search must be boolean; max_output_tokens must be 16..10000") requests, seen = [], set() for i, question in enumerate(questions): if not isinstance(question, str) or not question.strip() or len(question) > 10000: raise EvidenceError(f"Invalid question {i + 1}") normalized = folded(question) if normalized in seen: raise EvidenceError(f"Duplicate normalized question {i + 1}") seen.add(normalized) if any(folded(term) in normalized for term in terms): raise EvidenceError(f"Brand or alias leakage in question {i + 1}; no requests were made") payload = {"model": model, "input": question.strip(), "store": False, "max_output_tokens": tokens} if web: payload.update(tools=[{"type": "web_search"}], include=["web_search_call.action.sources"]) requests.append({"id": f"q{i + 1:03d}", "payload": payload}) return {"schema": "gptfav.plan.v1", "version": VERSION, "created_at": now(), "local_config": copy.deepcopy(config), "requests": requests, "notes": ["Local brand terms are never included in outgoing request payloads", "Each request has a single input and no previous_response_id, conversation or history", "Lexical leakage checks cannot detect unknown aliases, indirect brand hints or semantic leakage", "This is a Responses API plan, not a ChatGPT webpage new-chat experiment; model access and costs remain unverified"]} def validated_requests(plan): if not isinstance(plan, dict) or plan.get("schema") != "gptfav.plan.v1": raise EvidenceError("Expected a gptfav.plan.v1 file") expected = make_plan(plan.get("local_config"))["requests"] if plan.get("requests") != expected: raise EvidenceError("Plan requests were modified; regenerate from the local config") return expected def api_transport(payload): key = os.environ.get("OPENAI_API_KEY") if not key: raise EvidenceError("OPENAI_API_KEY must be supplied through the environment for --execute") request = urllib.request.Request(ENDPOINT, data=json.dumps(payload).encode("utf-8"), headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"}, method="POST") # Refuse redirects so authorization cannot be forwarded to another endpoint. class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None opener = urllib.request.build_opener(NoRedirect()) with opener.open(request, timeout=45) as response: body = response.read(MAX_BYTES + 1) if len(body) > MAX_BYTES: raise EvidenceError("API response exceeds 16 MiB") return json.loads(body) def run_plan(plan, output, *, execute=False, transport=None, max_requests=20): requests = validated_requests(plan) if type(max_requests) is not int or not 1 <= max_requests <= 100 or len(requests) > max_requests: raise EvidenceError("Plan exceeds the explicit request limit") if execute and transport is None and not os.environ.get("OPENAI_API_KEY"): raise EvidenceError("Set OPENAI_API_KEY in the environment before using --execute") result = {"schema": "gptfav.run.v1", "version": VERSION, "started_at": now(), "mode": "execute" if execute else "dry-run", "planned_requests": copy.deepcopy(requests), "records": [], "network_execution_requested": execute} write_json(output, result) if not execute: return result send = transport or api_transport for request in requests: entry = {"question_id": request["id"], "started_at": now(), "payload": copy.deepcopy(request["payload"]), "status": "failed"} try: raw = send(copy.deepcopy(request["payload"])) entry["raw_response"] = raw entry["evidence"] = extract_response(raw) if raw.get("status") == "completed": entry["status"] = "completed" else: entry["error"] = {"type": "IncompleteResponse", "response_status": raw.get("status")} except Exception as exc: # Never persist exception messages, response headers or authorization. entry["error"] = {"type": type(exc).__name__} if isinstance(exc, urllib.error.HTTPError): entry["error"]["http_status"] = exc.code entry["finished_at"] = now() result["records"].append(entry) write_json(output, result, replace=True) result["finished_at"] = now() write_json(output, result, replace=True) return result def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) extraction = commands.add_parser("extract", help="Extract explicit body citations; never fetch URLs") extraction.add_argument("input") extraction.add_argument("--output", required=True) extraction.add_argument("--mention-term", action="append", default=[]) report = commands.add_parser("report", help="Group citations by actual page URL") report.add_argument("inputs", nargs="+") report.add_argument("--output", required=True) plan = commands.add_parser("plan", help="Validate brand-free independent request payloads") plan.add_argument("config") plan.add_argument("--output", required=True) run = commands.add_parser("run", help="Dry run by default; --execute may incur API charges") run.add_argument("plan") run.add_argument("--output", required=True) run.add_argument("--execute", action="store_true") run.add_argument("--max-requests", type=int, default=20) args = parser.parse_args(argv) try: if args.command == "extract": value = extract_response(read_json(args.input), args.mention_term) elif args.command == "report": value = report_evidence([read_json(path) for path in args.inputs]) elif args.command == "plan": value = make_plan(read_json(args.config)) else: value = run_plan(read_json(args.plan), args.output, execute=args.execute, max_requests=args.max_requests) if args.command != "run": write_json(args.output, value) print(json.dumps({"command": args.command, "output": args.output, "version": VERSION, "mode": value.get("mode", "offline")}, ensure_ascii=False)) return 0 except (EvidenceError, OSError, ValueError) as exc: parser.exit(2, f"Error: {exc}\n") if __name__ == "__main__": main()