"""Source certification (Layer 1, spec 7). A certificate records source identity (size, mtime_ns), the full-file SHA-256 when actually computed, and format facts sniffed from the first 1 MiB / first 1000 parseable rows. The FULLY_VERIFIED status is written only when the full hash was actually computed; the timeout path records VERIFIED_WITH_LIMITATION with sha256_mode="not_computed" and never claims a full hash (G-3). A size/mtime change detected mid-read is automatic FAILED. """ import os import time from . import versions from .util import atomic_write_json, canonical_json, sha256_bytes from .parse import split_lines, header_keyword from .chunkmap import iter_line_spans SNIFF_BYTES = 1024 * 1024 SNIFF_ROWS = 1000 TS_DOTTED_MS_RE = "dotted_ms" TS_DOTTED_S_RE = "dotted_s" def _classify_line_endings(data): crlf = lf = cr = 0 i, n = 0, len(data) while i < n: b = data[i] if b == 0x0A: lf += 1 i += 1 elif b == 0x0D: if i + 1 < n and data[i + 1] == 0x0A: crlf += 1 i += 2 else: cr += 1 i += 1 else: i += 1 total = crlf + lf + cr if total == 0: return "LF", {"crlf": 0, "lf": 0, "cr": 0} which = "MIXED" if crlf and not lf and not cr: which = "CRLF" elif lf and not crlf and not cr: which = "LF" elif cr and not crlf and not lf: which = "CR" return which, {"crlf": crlf, "lf": lf, "cr": cr} def _detect_encoding(data): try: data.decode("ascii") return "ascii" except UnicodeDecodeError: try: data.decode("utf-8") return "utf8" except UnicodeDecodeError: return "other" def _is_compact(date_field, time_field): import re return (re.match(r"^\d{8}$", date_field) is not None and re.match(r"^\d{2}:\d{2}:\d{2}$", time_field) is not None) def _is_dotted_ms(f): import re return re.match(r"^\d{4}\.\d{2}\.\d{2} \d{2}:\d{2}:\d{2}\.\d{3}$", f) is not None def _is_dotted_s(f): import re return re.match(r"^\d{4}\.\d{2}\.\d{2} \d{2}:\d{2}:\d{2}$", f) is not None def _detect_format_rows(data): """Sniff up to SNIFF_ROWS parseable rows; return observed facts.""" rows = 0 field_count_hist = {} ts_formats = {} all_fields_comma = True has_header = False first_row_seen = False for line_bytes, _term in split_lines(data): if not line_bytes: continue if rows >= SNIFF_ROWS: break try: text = line_bytes.decode("utf-8") except UnicodeDecodeError: break fields = text.split(",") if len(fields) < 2: all_fields_comma = False continue field_count_hist[len(fields)] = field_count_hist.get(len(fields), 0) + 1 if not first_row_seen: first_row_seen = True if header_keyword(fields[0]): has_header = True continue # header row does not count as a data row ts_f = fields[0] if len(fields) >= 2 and _is_compact(fields[0], fields[1]): ts_formats["compact_date_time"] = ts_formats.get("compact_date_time", 0) + 1 elif _is_dotted_ms(ts_f): ts_formats["dotted_ms"] = ts_formats.get("dotted_ms", 0) + 1 elif _is_dotted_s(ts_f): ts_formats["dotted_s"] = ts_formats.get("dotted_s", 0) + 1 else: ts_formats["unrecognized"] = ts_formats.get("unrecognized", 0) + 1 rows += 1 return { "rows_sniffed": rows, "field_count_hist": field_count_hist, "ts_formats": ts_formats, "first_field_comma_split": all_fields_comma, "has_header": has_header, } def certify_source(source_path, output_path, certify_timeout_sec=3600, block=8 * 1024 * 1024): """Produce the source certificate at ``output_path`` (atomic).""" st_pre = os.stat(source_path) size = st_pre.st_size mtime_ns = st_pre.st_mtime_ns started = time.time() hasher = __import__("hashlib").new("sha256") sniffed = b"" rows = None fully_read = True status = "FULLY_VERIFIED" with open(source_path, "rb") as fh: while True: if time.time() - started > certify_timeout_sec: fully_read = False status = "VERIFIED_WITH_LIMITATION" break chunk = fh.read(block) if not chunk: break hasher.update(chunk) if len(sniffed) < SNIFF_BYTES: sniffed += chunk[: SNIFF_BYTES - len(sniffed)] elapsed = time.time() - started st_post = os.stat(source_path) if st_post.st_size != size or st_post.st_mtime_ns != mtime_ns: cert = _build_cert(source_path, None, size, mtime_ns, st_post, "FAILED", "source_changed_during_certify", None) atomic_write_json(output_path, cert) return cert if status == "VERIFIED_WITH_LIMITATION": cert = _build_cert(source_path, None, size, mtime_ns, st_post, "VERIFIED_WITH_LIMITATION", "timeout", None, sniffed=sniffed) atomic_write_json(output_path, cert) return cert full_sha = hasher.hexdigest() cert = _build_cert(source_path, full_sha, size, mtime_ns, st_post, "FULLY_VERIFIED", None, elapsed, sniffed=sniffed) atomic_write_json(output_path, cert) return cert def _build_cert(source_path, full_sha, size, mtime_ns, st_post, status, limitation_reason, elapsed, sniffed=b""): le_class, le_ratio = _classify_line_endings(sniffed) enc = _detect_encoding(sniffed) facts = _detect_format_rows(sniffed) ts_formats = facts["ts_formats"] ts_detected = "unrecognized" if ts_formats.get("dotted_ms"): ts_detected = "dotted_ms" elif ts_formats.get("dotted_s"): ts_detected = "dotted_s" elif ts_formats.get("compact_date_time"): ts_detected = "compact_date_time" field_counts = facts["field_count_hist"] observed_fields = max(field_counts, key=field_counts.get) if field_counts else 0 names = ["datetime", "bid", "ask"] if observed_fields == 4: names.append("vol") elif observed_fields == 6: names = ["date", "time", "bid", "ask", "last", "volume"] delim = "comma" if facts["first_field_comma_split"] or field_counts else "unrecognized" # Source-format decision (P3-DE-004 amendment). Two certified grammar # presets exist; the certificate pins grammar_id so init can cross-check # the configured grammar (fail-closed, never silent). source_format = "unrecognized" grammar_id = None if field_counts and observed_fields in (3, 4) and ts_detected in ("dotted_ms", "dotted_s"): source_format = "csv_tickstory_mt5" grammar_id = versions.GRAMMAR_ID_DOTTED elif field_counts and observed_fields == 6 and ts_detected == "compact_date_time": source_format = "csv_tickstory_mt5" grammar_id = versions.GRAMMAR_ID_TICKSTORY elif field_counts: source_format = "mismatch" cert = { "schema_version": versions.CERT_SCHEMA_VERSION, "certification_timestamp": _utc_now_iso(), "certification_status": status, "elapsed_sec": round(elapsed, 3) if elapsed is not None else None, "source_path": os.path.abspath(source_path), "source_size_bytes": size, "source_mtime_ns": mtime_ns, "sha256_full": full_sha, "sha256_mode": "full" if full_sha else "not_computed", "limitation_reason": limitation_reason, "source_format": source_format, "grammar_id": grammar_id, "line_ending": le_class, "line_ending_ratio": le_ratio, "line_ending_total": sum(le_ratio.values()), "encoding": enc, "timestamp_format_detected": ts_detected, "delimiter": delim, "has_header": bool(facts["has_header"]), "column_layout": {"fields": observed_fields, "names": names}, "parser_version": versions.PARSER_VERSION, "engine_version": versions.ENGINE_VERSION, } return cert def _utc_now_iso(): import datetime return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "+00:00") def cert_matches_source(cert, source_path): """Resume guard: size+mtime recomputed vs certificate (spec 13.5 #2).""" st = os.stat(source_path) if cert.get("source_size_bytes") != st.st_size: return "source_size_changed" if cert.get("source_mtime_ns") != st.st_mtime_ns: return "source_mtime_changed" return None