"""Persistent log storage for Squid access.log. This module owns the bridge between raw access.log files on disk and the SQLite-backed :class:`~app.LogEntry` table. It exposes: - :func:`sync_log_to_db` - incrementally import new lines from a log file - :func:`query_logs` - SQL-backed filter/search with time range support - :func:`get_log_stats` - aggregated statistics (hit/miss, top clients...) - :func:`get_distinct_hosts` / :func:`get_distinct_clients` / :func:`get_distinct_methods` - faceted browsing Implementation notes: - We rely on the existing :mod:`log_parser` for line format handling so we do not duplicate the access.log grammar. - Bulk inserts are batched (default 500 rows) using :func:`sqlalchemy.orm.Session.execute` with a parameterised INSERT so we don't pay the round-trip cost of per-row commits. - De-duplication uses the tuple ``(instance_id, log_time, client, raw_line)``; on conflict (we surface those as ``skipped_count`` rather than re-inserting). """ from __future__ import annotations import os import time from collections import Counter, defaultdict from datetime import datetime, timezone from typing import Iterable from sqlalchemy import and_, func, or_, select from sqlalchemy.exc import SQLAlchemyError import log_parser from app import db, LogEntry # imported lazily-friendly via app # --------------------------------------------------------------------------- # Sync # --------------------------------------------------------------------------- def _iter_log_lines(path: str, max_lines: int | None = None): """Yield raw lines from ``path`` as bytes-decoded strings. When ``max_lines`` is set we walk backwards through the file with a bounded per-iteration buffer (Deque-style window) and yield lines in chronological order. For an unbounded scan (full-history import) we just stream forwards. Designed to be tolerant of mid-stream truncation or binary garbage. """ if not os.path.isfile(path): return if not max_lines or max_lines <= 0: try: with open(path, "r", encoding="utf-8", errors="replace") as fh: for raw in fh: line = raw.rstrip("\n").rstrip("\r") if line.strip(): yield line except OSError: return return # Bounded tail: read last ``max_lines`` lines efficiently try: with open(path, "rb") as fh: try: fh.seek(0, os.SEEK_END) size = fh.tell() chunks: list[bytes] = [] lines_found = 0 block = 65536 pos = size while pos > 0 and lines_found <= max_lines: read_size = min(block, pos) pos -= read_size fh.seek(pos) chunks.append(fh.read(read_size)) lines_found += chunks[-1].count(b"\n") data = b"".join(reversed(chunks)) except OSError: fh.seek(0) data = fh.read() text = data.decode("utf-8", errors="replace") lines = text.splitlines() if len(lines) > max_lines: lines = lines[-max_lines:] for line in lines: stripped = line.strip() if stripped: yield stripped except OSError: return def sync_log_to_db( path: str, instance_id: int = 0, batch_size: int = 500, max_lines: int = 100000, ) -> tuple[int, int, str | None]: """Incrementally import new lines from ``path`` into :class:`LogEntry`. Steps: 1. Compute the high-water mark ``max(existing.log_time)`` for the instance so we can skip already-imported lines quickly. 2. Stream the last ``max_lines`` lines of the file (or the entire file when ``max_lines`` is None/0) and parse them through :func:`log_parser.parse_line`. 3. Insert new rows in batches of ``batch_size`` using :func:`sqlalchemy.dialects.sqlite.insert` with ``OR (primary key uniqueness handled by auot-increment)`` semantics. Duplicate suppression uses an in-memory ``seen`` set keyed by ``(client, log_time, raw_line)`` since the file itself can repeat a row at rotation boundaries. 4. Commit once per batch and return a tuple ``(inserted, skipped, err)``. All exceptions are caught and reported as the third tuple element so the caller can decide whether to surface them (e.g. on the dashboard). """ if batch_size <= 0: batch_size = 500 try: # high-water mark: only lines with log_time > wm are new wm = ( db.session.query(func.max(LogEntry.log_time)) .filter(LogEntry.instance_id == instance_id) .scalar() ) if wm is None: wm = 0.0 seen_keys: set[tuple[str, float, str]] = set() inserted = 0 skipped = 0 batch: list[dict] = [] for raw_line in _iter_log_lines(path, max_lines=max_lines): e = log_parser.parse_line(raw_line) if not e: skipped += 1 continue try: ts = float(e.get("time") or 0) except (TypeError, ValueError): skipped += 1 continue if ts <= wm: skipped += 1 continue key = (e.get("client") or "", ts, raw_line) if key in seen_keys: skipped += 1 continue seen_keys.add(key) batch.append({ "instance_id": instance_id, "log_time": ts, "elapsed_ms": int(e.get("elapsed_ms") or 0), "client": (e.get("client") or "")[:64], "result_code": (e.get("result_code") or "")[:32], "http_code": (e.get("http_code") or "")[:8], "size_bytes": int(e.get("size") or 0), "method": (e.get("method") or "")[:16], "url": (e.get("url") or "")[:65535], "host": (e.get("host") or "")[:256], "hier_code": (e.get("hier_code") or "")[:32], "content_type": (e.get("content_type") or "")[:128], "raw_line": raw_line[:65535], "created_at": datetime.now(timezone.utc), }) if len(batch) >= batch_size: _flush_batch(batch) inserted += len(batch) batch = [] if batch: _flush_batch(batch) inserted += len(batch) # record last sync time so the UI can display "上次同步" _record_last_sync(instance_id, path) return inserted, skipped, None except SQLAlchemyError as e: try: db.session.rollback() except Exception: pass return 0, 0, f"DB error: {type(e).__name__}: {e}" except Exception as e: try: db.session.rollback() except Exception: pass return 0, 0, f"{type(e).__name__}: {e}" def _column_count() -> int: """Return the number of columns in the :class:`~app.LogEntry` table. Used to cap the per-statement row count so the single INSERT does not exceed SQLite's 999-boundary-variable limit per SQL statement. """ return len(LogEntry.__table__.columns) # SQLite hard limit: at most 999 ? placeholders per single SQL statement. _SQLITE_MAX_VARIABLES = 999 def _max_rows_per_stmt() -> int: """Max rows that can safely fit in one parameterised INSERT.""" cols = _column_count() return max(1, (_SQLITE_MAX_VARIABLES - 1) // cols) def _flush_batch(batch: list[dict]) -> None: """Insert ``batch`` rows, splitting into chunks small enough for SQLite. SQLite rejects a single parameterised INSERT when its bound-variable count exceeds 999. Since :class:`~app.LogEntry` carries many columns, a 'large' batch (e.g. 500 rows) would blow past that limit. Split the batch into sub-chunks of ``_max_rows_per_stmt()`` rows and execute each sub-statement individually, then commit once the whole batch is in. """ if not batch: return from sqlalchemy.dialects.sqlite import insert as sqlite_insert chunk_size = _max_rows_per_stmt() for start in range(0, len(batch), chunk_size): chunk = batch[start:start + chunk_size] stmt = sqlite_insert(LogEntry).values(chunk) db.session.execute(stmt) db.session.commit() # --------------------------------------------------------------------------- # Query # --------------------------------------------------------------------------- def _parse_time(value) -> float | None: """Parse an ISO-ish datetime / unix timestamp into a unix float (UTC).""" if value is None or value == "": return None if isinstance(value, (int, float)): return float(value) if isinstance(value, datetime): if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) return value.timestamp() if isinstance(value, str): s = value.strip() if not s: return None # datetime-local HTML input sends strings like "2024-01-31T10:00" # or "2024-01-31 10:00:00". Python's fromisoformat handles both. try: # Support trailing Z if s.endswith("Z"): s = s[:-1] + "+00:00" dt = datetime.fromisoformat(s) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.timestamp() except ValueError: # fall through to float attempt pass try: return float(s) except ValueError: return None return None def _entry_to_dict(row: LogEntry) -> dict: """Convert a :class:`LogEntry` row to the dict shape used downstream. Mirrors :class:`log_parser.LogEntry` (which is a dict subclass) so existing consumers - templates, alert rules, anomaly detection - keep working without modification. """ ts = float(row.log_time or 0) host = row.host or "" url = row.url or "" raw = row.raw_line or "" # split result_code (already stored without slash) - reconstruct full # Squid "result" string (e.g. "TCP_HIT/200") for backward compat result_full = row.result_code or "" if row.http_code: result_full = f"{result_full}/{row.http_code}" return { "id": row.id, "time": row.log_time, # float unix ts "timestamp": datetime.fromtimestamp(ts, tz=timezone.utc) if ts > 0 else None, "elapsed_ms": int(row.elapsed_ms or 0), "client": row.client or "", "result": result_full, "result_code": row.result_code or "", "http_code": row.http_code or "", "category": log_parser.RESULT_CATEGORIES.get(row.result_code or "", "Other"), "result_description": log_parser.RESULT_CODE_DESCRIPTION.get(row.result_code or "", row.result_code or ""), "size": int(row.size_bytes or 0), "method": row.method or "", "url": url, "host": host, "ident": "", # not persisted separately "hierarchy": row.hier_code or "", "hier_code": row.hier_code or "", "peer": "", # not persisted "content_type": row.content_type or "", "raw_line": raw, "formatted_size": log_parser.format_bytes(int(row.size_bytes or 0)), "formatted_duration": log_parser.format_duration(int(row.elapsed_ms or 0)), } def query_logs(filters: dict) -> list[dict]: """Run a filtered SQL query against the persisted log rows. Supported keys (all optional): - ``instance_id`` (int) - ``client``, ``method``, ``result_code``, ``host`` (exact or substring) - ``host_filter_mode`` "exact" | "contains" (default "contains") - ``url`` substring LIKE - ``min_size``, ``max_size`` (int bytes) - ``min_elapsed``, ``max_elapsed`` (int milliseconds) - ``start_time``, ``end_time`` (datetime or unix timestamp, inclusive lo, inclusive hi for end so users don't accidentally exclude the minute they typed) - ``limit`` (default 1000), ``offset`` (default 0) Returns a list of dicts with the same shape as :func:`log_parser.parse_lines`, plus ``formatted_size`` / ``formatted_duration`` keys. """ try: q = db.session.query(LogEntry) # instance scoping iid = filters.get("instance_id") if iid is not None: q = q.filter(LogEntry.instance_id == iid) c = (filters.get("client") or "").strip() if c: q = q.filter(LogEntry.client.like(f"%{c}%")) m = (filters.get("method") or "").strip() if m: q = q.filter(LogEntry.method == m) rc = (filters.get("result_code") or "").strip() if rc: q = q.filter(LogEntry.result_code.like(f"%{rc}%")) h = (filters.get("host") or "").strip() if h: mode = (filters.get("host_filter_mode") or "contains").lower() if mode == "exact": q = q.filter(LogEntry.host == h) else: q = q.filter(LogEntry.host.like(f"%{h}%")) u = (filters.get("url") or "").strip() if u: q = q.filter(LogEntry.url.like(f"%{u}%")) mn = filters.get("min_size") if mn is not None: q = q.filter(LogEntry.size_bytes >= int(mn)) mx = filters.get("max_size") if mx is not None: q = q.filter(LogEntry.size_bytes <= int(mx)) mne = filters.get("min_elapsed") if mne is not None: q = q.filter(LogEntry.elapsed_ms >= int(mne)) mxe = filters.get("max_elapsed") if mxe is not None: q = q.filter(LogEntry.elapsed_ms <= int(mxe)) st = _parse_time(filters.get("start_time")) if st is not None: q = q.filter(LogEntry.log_time >= st) et = _parse_time(filters.get("end_time")) if et is not None: q = q.filter(LogEntry.log_time <= et) # default ordering: newest first q = q.order_by(LogEntry.log_time.desc(), LogEntry.id.desc()) limit = int(filters.get("limit") or 1000) offset = int(filters.get("offset") or 0) if limit > 0: q = q.limit(limit) if offset > 0: q = q.offset(offset) rows = q.all() return [_entry_to_dict(r) for r in rows] except SQLAlchemyError as e: try: db.session.rollback() except Exception: pass return [] def count_logs(filters: dict | None = None) -> int: """Return the number of rows matching ``filters`` (same shape as ``query_logs``). Used by ``logs_view`` to render total / pagination accurately without transferring every row over the wire. """ f = dict(filters or {}) f["limit"] = 0 f["offset"] = 0 try: # rebuild only the WHERE portion - run the same filter logic but # count instead of fetch q = _build_base_query(f) return q.with_entities(func.count(LogEntry.id)).scalar() or 0 except SQLAlchemyError: try: db.session.rollback() except Exception: pass return 0 def _build_base_query(filters: dict): """Internal: same WHERE-clause as :func:`query_logs` but returns the SQLAlchemy query for further composition (e.g. count).""" q = db.session.query(LogEntry) iid = filters.get("instance_id") if iid is not None: q = q.filter(LogEntry.instance_id == iid) c = (filters.get("client") or "").strip() if c: q = q.filter(LogEntry.client.like(f"%{c}%")) m = (filters.get("method") or "").strip() if m: q = q.filter(LogEntry.method == m) rc = (filters.get("result_code") or "").strip() if rc: q = q.filter(LogEntry.result_code.like(f"%{rc}%")) h = (filters.get("host") or "").strip() if h: q = q.filter(LogEntry.host.like(f"%{h}%")) u = (filters.get("url") or "").strip() if u: q = q.filter(LogEntry.url.like(f"%{u}%")) mn = filters.get("min_size") if mn is not None: q = q.filter(LogEntry.size_bytes >= int(mn)) mx = filters.get("max_size") if mx is not None: q = q.filter(LogEntry.size_bytes <= int(mx)) mne = filters.get("min_elapsed") if mne is not None: q = q.filter(LogEntry.elapsed_ms >= int(mne)) mxe = filters.get("max_elapsed") if mxe is not None: q = q.filter(LogEntry.elapsed_ms <= int(mxe)) st = _parse_time(filters.get("start_time")) if st is not None: q = q.filter(LogEntry.log_time >= st) et = _parse_time(filters.get("end_time")) if et is not None: q = q.filter(LogEntry.log_time <= et) return q # --------------------------------------------------------------------------- # Aggregate stats # --------------------------------------------------------------------------- def get_log_stats( instance_id: int = 0, start_time: float | None = None, end_time: float | None = None, ) -> dict: """Compute aggregate statistics in SQL. Compatible with :func:`log_parser.aggregate_stats`. When there are no rows in scope we return the same ``_empty_stats()`` structure so the caller doesn't need a special case. """ try: q = _build_base_query({ "instance_id": instance_id, "start_time": start_time, "end_time": end_time, }) total = q.with_entities(func.count(LogEntry.id)).scalar() or 0 if total == 0: stats = _copy_empty_stats() else: agg = q.with_entities( func.coalesce(func.sum(LogEntry.size_bytes), 0), func.coalesce(func.sum(LogEntry.elapsed_ms), 0), func.min(LogEntry.log_time), func.max(LogEntry.log_time), ).one() total_bytes, total_elapsed, ts_min, ts_max = agg total = int(total) total_bytes = int(total_bytes) total_elapsed = int(total_elapsed) avg_elapsed = round(total_elapsed / total, 2) if total else 0.0 avg_size = round(total_bytes / total, 2) if total else 0.0 # result_code distribution by_result = dict( q.with_entities( LogEntry.result_code, func.count(LogEntry.id), ).group_by(LogEntry.result_code).all() ) by_result = {k or "": v for k, v in by_result.items()} # http_code distribution (only rows with a value) by_http = dict( q.filter(LogEntry.http_code != "").with_entities( LogEntry.http_code, func.count(LogEntry.id), ).group_by(LogEntry.http_code).all() ) # method distribution by_method = dict( q.with_entities( LogEntry.method, func.count(LogEntry.id), ).group_by(LogEntry.method).all() ) # hierarchy distribution by_hier = dict( q.with_entities( LogEntry.hier_code, func.count(LogEntry.id), ).group_by(LogEntry.hier_code).all() ) # top clients: most active + bytes top_clients = q.with_entities( LogEntry.client, func.count(LogEntry.id).label("c"), func.coalesce(func.sum(LogEntry.size_bytes), 0).label("b"), ).group_by(LogEntry.client).order_by(func.count(LogEntry.id).desc()).limit(50).all() top_clients = [{"client": c or "", "count": int(c_), "bytes": int(b_)} for c, c_, b_ in top_clients] # top hosts top_hosts_rows = q.filter(LogEntry.host != "").with_entities( LogEntry.host, func.count(LogEntry.id).label("c"), func.coalesce(func.sum(LogEntry.size_bytes), 0).label("b"), ).group_by(LogEntry.host).order_by(func.count(LogEntry.id).desc()).limit(50).all() top_hosts = [{"host": h, "count": int(c_), "bytes": int(b_)} for h, c_, b_ in top_hosts_rows] # top urls top_urls_rows = q.with_entities( LogEntry.url, func.count(LogEntry.id).label("c"), ).group_by(LogEntry.url).order_by(func.count(LogEntry.id).desc()).limit(50).all() top_urls = [{"url": u, "count": int(c_)} for u, c_ in top_urls_rows] # content-type: simpler approach - just count distinct top values by_ctype_rows = q.with_entities( LogEntry.content_type, func.count(LogEntry.id).label("c"), ).group_by(LogEntry.content_type).order_by(func.count(LogEntry.id).desc()).limit(20).all() by_ctype = [] for ct, c_ in by_ctype_rows: ct_main = (ct or "-").split(";")[0].strip() or "-" by_ctype.append((ct_main, int(c_))) # collapse same main types merged: dict[str, int] = defaultdict(int) for ct_main, c_ in by_ctype: merged[ct_main] += c_ by_ctype = list(merged.items())[:20] # hit/miss ratio using result_code categories hits = sum(c for code, c in by_result.items() if log_parser.RESULT_CATEGORIES.get(code or "") == "Hit") misses = sum(c for code, c in by_result.items() if log_parser.RESULT_CATEGORIES.get(code or "") == "Miss") denied = sum(c for code, c in by_result.items() if log_parser.RESULT_CATEGORIES.get(code or "") == "Denied") aborted = by_result.get("TCP_MISS_ABORTED", 0) denied_rate = denied / total if total else 0.0 aborted_rate = aborted / total if total else 0.0 hit_ratio = hits / (hits + misses) if (hits + misses) > 0 else 0.0 errors_4xx = sum(c for code, c in by_http.items() if code.startswith("4")) errors_5xx = sum(c for code, c in by_http.items() if code.startswith("5")) errors_4xx_rate = errors_4xx / total if total else 0.0 errors_5xx_rate = errors_5xx / total if total else 0.0 # unique clients / hosts unique_clients = q.with_entities(func.count(func.distinct(LogEntry.client))).scalar() or 0 unique_hosts = q.filter(LogEntry.host != "").with_entities( func.count(func.distinct(LogEntry.host)) ).scalar() or 0 # hourly distribution (UTC hour of day) by_hour_counts: dict[int, int] = defaultdict(int) by_hour_bytes: dict[int, int] = defaultdict(int) # we already have min/max so size of an in-Python bucket is small # but to avoid pulling all rows we can do an UNION-ish query per # hour; the simpler approach: fetch (time, size) projection. rows = q.with_entities(LogEntry.log_time, LogEntry.size_bytes).all() for ts, size in rows: if not ts: continue hr = datetime.fromtimestamp(float(ts), tz=timezone.utc).hour by_hour_counts[hr] += 1 by_hour_bytes[hr] += int(size or 0) time_start = datetime.fromtimestamp(float(ts_min), tz=timezone.utc) if ts_min else None time_end = datetime.fromtimestamp(float(ts_max), tz=timezone.utc) if ts_max else None stats = { "total_requests": total, "total_bytes": total_bytes, "total_elapsed_ms": total_elapsed, "unique_clients": int(unique_clients), "unique_hosts": int(unique_hosts), "hit_ratio": hit_ratio, "denied_rate": denied_rate, "aborted_rate": aborted_rate, "errors_4xx_rate": errors_4xx_rate, "errors_5xx_rate": errors_5xx_rate, "errors_4xx_count": errors_4xx, "errors_5xx_count": errors_5xx, "avg_elapsed_ms": avg_elapsed, "avg_size_bytes": avg_size, "hits": hits, "misses": misses, "denied": denied, "aborted": aborted, "time_start": time_start, "time_end": time_end, "by_result": log_parser._top_n(Counter(by_result), 50), "by_http": log_parser._top_n(Counter(by_http), 50), "by_method": log_parser._top_n(Counter(by_method), 20), "by_category": [], # we don't store category; compute inline if needed "by_hierarchy": log_parser._top_n(Counter(by_hier), 20), "by_ctype": by_ctype, "top_clients": top_clients, "top_hosts": top_hosts, "top_urls": top_urls, "hourly": [ {"hour": h, "requests": int(by_hour_counts.get(h, 0)), "bytes": int(by_hour_bytes.get(h, 0))} for h in range(24) ], "result_code_legend": log_parser.RESULT_CODE_DESCRIPTION, "hierarchy_legend": log_parser.HIERARCHY_LABELS, } # augment with sync timestamp for the dashboard UI try: from app import AppConfig ts_str = AppConfig.query.filter_by(key="_log_last_sync_at").first() stats["last_sync_at"] = ts_str.value if ts_str else None except Exception: stats["last_sync_at"] = None return stats except SQLAlchemyError as e: try: db.session.rollback() except Exception: pass s = _copy_empty_stats() s["last_sync_at"] = None s["error"] = f"{type(e).__name__}: {e}" return s def _copy_empty_stats() -> dict: """Return a fresh empty-stats dict (avoid mutating log_parser._empty_stats).""" return log_parser._empty_stats() # --------------------------------------------------------------------------- # Distinct facets # --------------------------------------------------------------------------- def get_distinct_hosts(instance_id: int = 0, limit: int = 200) -> list[str]: try: rows = ( db.session.query(LogEntry.host) .filter(LogEntry.host != "", LogEntry.instance_id == instance_id) .distinct() .order_by(LogEntry.host.asc()) .limit(int(limit)) .all() ) return [r[0] for r in rows if r[0]] except SQLAlchemyError: try: db.session.rollback() except Exception: pass return [] def get_distinct_clients(instance_id: int = 0, limit: int = 200) -> list[str]: try: rows = ( db.session.query(LogEntry.client) .filter(LogEntry.client != "", LogEntry.instance_id == instance_id) .distinct() .order_by(LogEntry.client.asc()) .limit(int(limit)) .all() ) return [r[0] for r in rows if r[0]] except SQLAlchemyError: try: db.session.rollback() except Exception: pass return [] def get_distinct_methods(instance_id: int = 0) -> list[str]: try: rows = ( db.session.query(LogEntry.method) .filter(LogEntry.method != "", LogEntry.instance_id == instance_id) .distinct() .order_by(LogEntry.method.asc()) .all() ) return [r[0] for r in rows if r[0]] except SQLAlchemyError: try: db.session.rollback() except Exception: pass return [] # --------------------------------------------------------------------------- # Sync bookkeeping # --------------------------------------------------------------------------- _LAST_SYNC_KEY = "_log_last_sync_at" def _record_last_sync(instance_id: int, path: str) -> None: """Stamp ``AppConfig`` rows with the last successful sync timestamp. Two values are stored: ``_log_last_sync_at`` (ISO timestamp) and ``_log_last_sync_path`` (the file we synced from). We use ``AppConfig`` rather than ``LogEntry`` so the timestamp survives even if log rows are later rotated out. """ try: from app import AppConfig ts_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") for key, value in ( (_LAST_SYNC_KEY, ts_iso), ("_log_last_sync_path", path or ""), ("_log_last_sync_instance", str(instance_id)), ): row = AppConfig.query.filter_by(key=key).first() if row is None: row = AppConfig(key=key, value=value) db.session.add(row) else: row.value = value db.session.commit() except Exception: try: db.session.rollback() except Exception: pass def get_last_sync_at() -> str | None: """Return the ISO timestamp of the last sync, or ``None``.""" try: from app import AppConfig row = AppConfig.query.filter_by(key=_LAST_SYNC_KEY).first() return row.value if row else None except Exception: return None def get_last_sync_path() -> str | None: try: from app import AppConfig row = AppConfig.query.filter_by(key="_log_last_sync_path").first() return row.value if row else None except Exception: return None def get_last_sync_at_local() -> str | None: """Return the last-sync timestamp formatted in local time (not UTC). :class:`~app.AppConfig` stores the sync time as an ISO-8601 UTC string (e.g. ``2026-07-16T07:43:36Z``). This converts it to the system's local timezone so the UI shows a human-friendly time to the operator. """ ts = get_last_sync_at() if not ts: return None try: dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S") except Exception: return ts def import_full_history( path: str, instance_id: int = 0, batch_size: int = 500, progress_cb=None, ) -> tuple[int, int, str | None]: """Import the entire ``access.log`` from byte 0 (not just the tail). Used by the "Import history" button on the logs page. Calls :func:`sync_log_to_db` with a very high ``max_lines`` (50M) so we stream every line. Progress is reported via ``progress_cb(processed)`` so the caller can render a status indicator. """ try: # 50M lines ceiling - enough for multi-year logs; streaming means # we never actually load them all into memory. return sync_log_to_db( path=path, instance_id=instance_id, batch_size=batch_size, max_lines=50_000_000, ) except Exception as e: return 0, 0, f"{type(e).__name__}: {e}"