API Reference

def backfill(start, end, body_id):
    """Extract old documents and fill db with summaries

    Args:
        start (str): start date in ISO format
        end (str): end date in ISO format
        body_id (int): numerical ID of the government body to filter for (In Windsor Town Council is 18)

    Returns:
        assignments (lis): list of assignment data in the same format as get_rss() output.
    """
    assignments = []
    base_url = "https://www.windsorct.gov/AgendaCenter/ViewFile/Agenda/_"
    start_date = datetime.date.fromisoformat(start)
    end_date = datetime.date.fromisoformat(end)
    # construct url and issue requests.get() 
    url = f"https://www.windsorct.gov/AgendaCenter/Search/?term=&CIDs={body_id},&startDate={start_date.month:02d}/{start_date.day:02d}/{start_date.year}&endDate={end_date.month:02d}/{end_date.day:02d}/{end_date.year}&dateRange=&dateSelector="
    # HTTP request
    response = requests.get(url, headers=HEADERS)

    if response.status_code != 200:
        logger.error("Backfill request failed: %s", response.status_code)
        return []

    time.sleep(REQUEST_DELAY)
    soup = BeautifulSoup(response.text, 'html.parser')
    leads = soup.select('td p a[id]')
    for lead in leads:
        body = classify(lead.text, MUNICIPAL_BODIES)
        body = "_".join(body.lower().split())
        meeting_type = classify(lead.text, MEETING_TYPES)
        month = lead["id"][0:2]
        day = lead["id"][2:4]
        year = lead["id"][4:8]
        published_date = f"{year}-{month}-{day}"
        meeting_id = f"{body}_{lead['name']}_{year}"
        materials = f"{base_url}{lead['id']}?html=true"

        assignments.append({
            "meeting_id": meeting_id,
            "meeting_type": meeting_type,
            "body": body,
            "published_date": published_date,
            "materials": materials
        })

    return assignments

def get_rss(rss_url):
    """Retrieves new meeting info from Town RSS feed

    Args:
        rss_url (str): hyperlink to RSS feed

    Raises:
        Exception: when HTTP response != 200

    Returns:
        new_entries (lis): list of new entries, each entry is a flat dict with keys {
                "meeting_id": composite_id,
                "meeting_type": meeting_type,
                "body": body,
                "published_date": pub_date,
                "materials": url
                }
    """

    feed = feedparser.parse(rss_url, agent=USER_AGENT, modified=None)
    civic_log.record(datetime.datetime.now(), rss_url, feed.status)

    if feed.status == 304:
        logger.info("STATUS: %s; No updates", feed.status)
        return {}

    if feed.status != 200:
        raise Exception("Feed error: %s", feed.status)

    # Parse the feed
    logger.info("STATUS: %s; Parsing new data", feed.status)
    new_entries = []
    for entry in feed.entries:
        try:    
            id_parts = entry["id"].split("/")
            meeting_id = id_parts[-2]
            title = entry["title"]
            body = classify(title, MUNICIPAL_BODIES)
            body = "_".join(body.lower().split())
            meeting_type = classify(title, MEETING_TYPES, meeting_type_fallback=True)
            year = entry["published_parsed"][0]
            month = entry["published_parsed"][1]
            day = entry["published_parsed"][2]
            pub_date = datetime.date(year, month, day).isoformat()

            composite_id = f"{body}_{meeting_id}_{year}"
            url = (
                f"https://www.windsorct.gov/AgendaCenter/"
                f"ViewFile/Agenda/_{month:02d}{day:02d}{year}-{meeting_id}?html=true"
            )

            new_entries.append({
                "meeting_id": composite_id,
                "meeting_type": meeting_type,
                "body": body,
                "published_date": pub_date,
                "materials": url
                })

        except KeyError as e:
            logger.warning("bad entry: %s", e)
            continue

    return new_entries

def approve_article(meeting_id, reviewed=True):
    """toggles reviewed boolean on an assignment"""
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("""UPDATE assignments SET reviewed = %s
                    WHERE meeting_id = %s""",
                    (reviewed, meeting_id))
        conn.commit()
    except Exception as e:
        conn.rollback()
        logger.warning(f"{e}")
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

def assign():
    """Review db for assignments with status=STATUS_PENDING and return them

    Returns:
        assignments (lis): list of pending assignments to be sent to reporter
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
        cur.execute(
            """SELECT assignments.meeting_id, meeting_type, materials, status
            FROM assignments
            LEFT JOIN articles ON assignments.meeting_id = articles.meeting_id
            WHERE assignments.status = 'pending' AND articles.meeting_id IS NULL;
            """
        )
        assignments = cur.fetchall()
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

    for assignment in assignments:
        update_status(assignment['meeting_id'], STATUS_ASSIGNED)

    return assignments

def publish_article(meeting_id, published=True):
    """toggles published boolean on an assignment"""
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("""UPDATE assignments SET published = %s
                    WHERE meeting_id = %s""",
                    (published, meeting_id))
        conn.commit()
    except Exception as e:
        conn.rollback()
        logger.warning(f"{e}")
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

def report_error(meeting_id, error_text):
    """inserts a new row to error_reports

    Args:
        meeting_id (str): composit meeting id
        error_text (str): error message from the POST '/report-error' route
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("""INSERT INTO error_reports (meeting_id, report_text) VALUES (%s, %s)""",
                    (meeting_id, error_text))
        conn.commit()
    except Exception as e:
        conn.rollback()
        logger.warning(f"{e}")
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

def save_ai_log(logs):
    """save outgoing request logs to ai providers

    Args:
        logs (lis): list of tuples returned by ai_log.info
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        for log in logs:
            try:
                cur.execute("""
                            INSERT INTO ai_requests (run_id, timestamp, provider, status, input_tokens, output_tokens)
                            VALUES (%s, %s, %s, %s, %s, %s)""",
                            (log[0], log[1], log[2], log[3], log[4], log[5]))
                conn.commit()
            except Exception as e:
                conn.rollback()
                logger.warning(f"{e}")
                continue
    finally:
        if cur: cur.close()
        if conn: conn.close()

def save_articles(articles):
    """Recieve articles from writer.write_article() and add them to the articles table

    Args:
        articles (lis): list of dicts containing article data ("headline", "summary", etc...)
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        for article in articles:
            cur.execute(
                """
                INSERT INTO articles (meeting_id, meeting_date, byline, doc_type, summary)
                VALUES  (%s, %s, %s, %s, %s)
                ON CONFLICT (meeting_id, doc_type) DO UPDATE SET
                    summary = EXCLUDED.summary,
                    meeting_date = EXCLUDED.meeting_date
                """,
                (article['meeting_id'], article['meeting_date'], article['byline'], article["doc_type"], json.dumps(article['summary']))
            )
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

def save_assignments(rss_assignments, run_id=None):
    """Add new assignments from intake.get_rss() to the assignments table

    Args:
        rss_assignments (lis): List of assignments produced by either intake.get_rss() or intake.backfill()
        run_id (int): global run id associated with the cron run
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        for assignment in rss_assignments:
            cur.execute(
                """INSERT INTO assignments (meeting_id, meeting_type, body, published_date, materials, last_run_id)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON CONFLICT (meeting_id) DO UPDATE SET
                    meeting_type=EXCLUDED.meeting_type,
                    materials=EXCLUDED.materials,
                    body=EXCLUDED.body,
                    published_date=EXCLUDED.published_date,
                    status='pending',
                    last_run_id=EXCLUDED.last_run_id
                WHERE assignments.materials != EXCLUDED.materials
                """, (assignment['meeting_id'], assignment['meeting_type'], assignment['body'], assignment['published_date'], assignment['materials'], run_id)
                )
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

def save_civic_log(logs):
    """save outgoing request logs to town servers

    Args:
        logs (lis): list of tuples returned by civic_log.info
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        for log in logs:
            try:
                cur.execute("""
                            INSERT INTO civic_requests (run_id, timestamp, url, response_status)
                            VALUES (%s, %s, %s, %s)""",
                            (log[0], log[1], log[2], log[3]))
                conn.commit()
            except Exception as e:
                conn.rollback()
                logger.warning(f"{e}")
                continue
    finally:    
        if cur: cur.close()
        if conn: conn.close()

def send_alert(error):
    """Email the developer when the wuiw.main routine fails

    Args:
        error (str): error message
    """
    sender = os.getenv("ALERT_EMAIL")
    password = os.getenv("ALERT_EMAIL_PASSWORD")
    recipient = os.getenv("ALERT_EMAIL")  # send to yourself

    msg = MIMEText(f"WUIW pipeline failed.\n\nError: {error}")
    msg["Subject"] = "WUIW Pipeline Failure"
    msg["From"] = sender
    msg["To"] = recipient

    try:
        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
            server.login(sender, password)
            server.sendmail(sender, recipient, msg.as_string())
        logger.info("Failure alert sent")
    except Exception as e:
        logger.error("Failed to send alert: %s", e)

def update_article(meeting_id, updates, resolved=False):
    """saves edits, sets reviewed = True, optionally resolves all open error reports
    updates is type dict and is constructed by the POST route when updates are submitted

    Args:
        meeting_id (str): composite meeting id (example: town_council_1234_2025)
        updates (dict): constructed by the form in the POST '/admin/articles/<meeting_id>' route
            {
            "assignment": {
                "meeting_type": "Regular Meeting"
            },
            "article": {
                "agenda": {
                    "items": ["items"]
                },
                "minutes": {
                    "meeting_date": "2025-03-01",
                    "headline": "headline",
                    "bullets": ["bullets"],
                    "blurb": "blurb"
                    }
                }
            }
        resolved (bool, optional): Toggles resolved status. Defaults to False.

    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        # update assignments.meeting_type
        cur.execute("""UPDATE assignments SET meeting_type = %s WHERE meeting_id = %s""",
                    (updates["assignment"]["meeting_type"], meeting_id))

        # update articles.meeting_date and articles.summary
        for doc_type, data in updates["article"].items():
            if doc_type == "agenda":
                pass

            if doc_type == "minutes":
                cur.execute("""UPDATE articles SET
                            summary = %s,
                            meeting_date = %s
                            WHERE meeting_id = %s AND doc_type = %s""",
                            (json.dumps(data), data["meeting_date"], meeting_id, doc_type))

            if doc_type == "voting_grid":
                pass

        # update assignments.reviewed
        cur.execute("""UPDATE assignments SET reviewed = TRUE
                    WHERE meeting_id = %s""",
                    (meeting_id,))

        # update error_reports.status
        if resolved:
            cur.execute("""UPDATE error_reports SET resolved = TRUE
                        WHERE meeting_id = %s""",
                        (meeting_id,))

        conn.commit()
    except Exception as e:
        conn.rollback()
        logger.warning(f"{e}")
        raise
    finally:
        if cur: cur.close()
        if conn: conn.close()

def update_status(meeting_id, status, error_message=None):
    """Update status of an assignment in the database

    Args:
        meeting_id (str): composite meeting id
        status (str): assignment status as prescribed in config
        error_message (str): message with info about errors that occured in this assignment upstream
    """
    conn = None
    cur = None
    try:
        conn = get_db_connection()
        cur = conn.cursor()
        if error_message:
            cur.execute("UPDATE assignments SET status = %s, error_message = %s where meeting_id=%s", (status, error_message, meeting_id))
        else:
            cur.execute("UPDATE assignments SET status = %s, error_message = NULL WHERE meeting_id = %s", (status, meeting_id))
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
       if cur: cur.close()
       if conn: conn.close()

def fetch_documents(url, doc_type=None):
    """Use beautiful soup to parse html for urls to pdf(s)

    Args:
        url (str): link to town documents
        doc_type (lis): specifies which type of docs to get

    Returns:
        documents (tup): tuple with items (documents list, status, error message)
    """
    response = requests.get(url, headers=HEADERS)
    civic_log.record(datetime.now(), url, response.status_code)
    time.sleep(REQUEST_DELAY)

    if response.status_code != 200:
        logger.warning(f"materials url returned {response.status_code}")
        return ({}, STATUS_FAILED, f"materials get returned {response.status_code}")

    documents = {}
    soup = BeautifulSoup(response.text, 'html.parser')
    items = soup.find_all('div', class_='item level1')
    print(f"found {len(items)} documents to parse")

    target_docs = {}
    for item in items:
        title = item.find('h1', class_='title').text.strip()
        detected_type = classify(title, DOCUMENT_TYPES, doc_type_fallback=True)
        href = item.find('a')['href']
        doc_url = href if href.startswith("http") else f"https://www.windsorct.gov{href}"
        target_docs[detected_type] = doc_url

    keys = [doc_type] if doc_type is not None else target_docs.keys()

    for key in keys:
        if key not in target_docs:
            logger.warning(f"doc_type {key} not in materials")
            continue

        response_pdf = requests.get(target_docs[key], headers=HEADERS)
        civic_log.record(datetime.now(), target_docs[key], response_pdf.status_code)
        time.sleep(REQUEST_DELAY)

        if response_pdf.status_code != 200:
            logger.warning(f"No pdf returned for {key}; status: {response_pdf.status_code}")
            continue

        pdf_stream = io.BytesIO(response_pdf.content)
        text = _transcribe_doc(pdf_stream)
        documents[key] = text

    return (documents, STATUS_ASSIGNED, None)

def review_article(draft):
    """Run checks on data coming back from AI provider

    Args:
        draft (dict): article information created by writer.write_article()
    """
    # check returned object is dictionary
    if not isinstance(draft, dict):
        return None, STATUS_FAILED, "draft is not a dict"

    # check all keys exist
    missing = REQUIRED_KEYS - draft.keys()
    if missing:
        return None, STATUS_FAILED, f"missing keys: {missing}"

    # check date format validation
    try:
        datetime.fromisoformat(draft["meeting_date"])
    except ValueError:
        logger.warning("date value: %s is not ISO format", draft["meeting_date"])
        return None, STATUS_FAILED, f"date: {draft['meeting_date']} not in format YYYY-MM-DD"

    # check bullets is a list etc
    if not isinstance(draft["bullets"], list):
        return None, STATUS_FAILED, "bullets is not a list"

    return draft, STATUS_COMPLETE, None

def write_article(meeting_id, text, doc_type):
    """Send document text to journalist for summarization.

    Args:
        meeting_id (str): composite meeting id
        text (str): document text
        doc_type (str): document type for prompt construction

    Returns:
        article (tup): (article dictionary item, status, error message)
    """
    try:
        provider = get_provider()
        draft, client_status, input_tokens, output_tokens = provider.summarize(text, doc_type)
        ai_log.record(datetime.now(), provider.model, client_status, input_tokens, output_tokens)
    except Exception as e:
        ai_log.record(datetime.now(), provider.model, "FAIL", None, None)
        return None, STATUS_FAILED, f"summarize failed: {e}"

    article, status, error = review_article(draft)

    if status == STATUS_FAILED:
        return None, STATUS_FAILED, error

    return {
        "meeting_id": meeting_id,
        "meeting_date": article.get("meeting_date"),
        "byline": provider.model,
        "doc_type": doc_type,
        "summary": article
    }, STATUS_COMPLETE, None