load_tv_programs_sqlite.py

import re
from datetime import datetime
from pathlib import Path
import sqlite3

DB_PATH = "tvguide.db"
INPUT_FILES = ["tv_programs_BBC.txt","tv_programs_Disc.txt","tv_programs_NatGeo.txt"]
RECORD_SEP = re.compile(r"^-{3,}\s*$")
KV_LINE     = re.compile(r"^\s*([^:]+)\s*:\s*(.*)\s*$")

DDL = """
PRAGMA journal_mode=WAL;
PRAGMA foreign_keys=ON;

CREATE TABLE IF NOT EXISTS programs (
  id            INTEGER PRIMARY KEY AUTOINCREMENT,
  title         TEXT,
  day_name      TEXT,
  air_date      TEXT,   -- store ISO date string 'YYYY-MM-DD'
  start_time    TEXT,   -- store ISO time string 'HH:MM:SS'
  end_time      TEXT,
  duration_min  INTEGER,
  channel       TEXT,
  link          TEXT,
  original_name TEXT,
  prod_year     INTEGER,
  description   TEXT,
  score_pct     INTEGER,
  genre         TEXT,
  source_file   TEXT
);

CREATE UNIQUE INDEX IF NOT EXISTS ux_program
  ON programs(channel, air_date, start_time, title);
"""

UPSERT = """
INSERT INTO programs
  (title, day_name, air_date, start_time, end_time, duration_min,
   channel, link, original_name, prod_year, description, score_pct, genre, source_file)
VALUES
  (:title, :day_name, :air_date, :start_time, :end_time, :duration_min,
   :channel, :link, :original_name, :prod_year, :description, :score_pct, :genre, :source_file)
ON CONFLICT(channel, air_date, start_time, title) DO UPDATE SET
  day_name=excluded.day_name,
  end_time=excluded.end_time,
  duration_min=excluded.duration_min,
  link=excluded.link,
  original_name=excluded.original_name,
  prod_year=excluded.prod_year,
  description=excluded.description,
  score_pct=excluded.score_pct,
  genre=excluded.genre,
  source_file=excluded.source_file;
"""

def parse_file(path: Path):
    items = []
    with path.open("r", encoding="utf-8") as f:
        block = []
        for line in f:
            if RECORD_SEP.match(line):
                if block:
                    rec = parse_block(block, path.name)
                    if rec: items.append(rec)
                    block = []
            else:
                if line.strip():
                    block.append(line.rstrip("\n"))
        if block:
            rec = parse_block(block, path.name)
            if rec: items.append(rec)
    return items

def parse_block(lines, source_file):
    d = {}
    for ln in lines:
        m = KV_LINE.match(ln)
        if m:
            key, val = m.group(1).strip(), m.group(2).strip()
            d[key] = val

    def to_date_str(s):
        if not s: return None
        try:
            return datetime.strptime(s, "%d.%m.%Y").strftime("%Y-%m-%d")
        except ValueError:
            return None

    def to_time_str(s):
        if not s: return None
        try:
            # normalize to HH:MM:SS
            return datetime.strptime(s, "%H:%M").strftime("%H:%M:%S")
        except ValueError:
            return None

    def first_int(s):
        if not s: return None
        m = re.search(r"\d+", s)
        return int(m.group(0)) if m else None

    return {
        "title":         d.get("Title") or None,
        "day_name":      d.get("Day") or None,
        "air_date":      to_date_str(d.get("Date")),           # TEXT
        "start_time":    to_time_str(d.get("Start Time")),     # TEXT
        "end_time":      to_time_str(d.get("End Time")),       # TEXT
        "duration_min":  first_int(d.get("Duration")),
        "channel":       d.get("Channel") or None,
        "link":          d.get("Link") or None,
        "original_name": d.get("Original Name") or None,
        "prod_year":     first_int(d.get("Year")),
        "description":   d.get("Description") or None,
        "score_pct":     first_int(d.get("Score")),
        "genre":         d.get("Genre") or None,
        "source_file":   source_file,
    }

def main():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("PRAGMA synchronous=NORMAL;")
    for stmt in filter(None, DDL.split(";")):
        s = stmt.strip()
        if s: conn.execute(s + ";")

    all_rows = []
    for fname in INPUT_FILES:
        p = Path(fname)
        if not p.exists():
            print(f"WARNING: {fname} not found, skipping.")
            continue
        all_rows.extend(parse_file(p))
    if not all_rows:
        print("No rows parsed. Nothing to insert.")
        return

    with conn:
        conn.executemany(UPSERT, all_rows)

    print(f"Inserted/updated {len(all_rows)} rows into {DB_PATH}.")
    conn.close()

if __name__ == "__main__":
    main()