diff --git a/README.md b/README.md index 1d23597..cfb08d7 100644 --- a/README.md +++ b/README.md @@ -168,6 +168,43 @@ Use the production frontend — the CloudFront or custom-domain URL. Response quality improves as more documents finish ingestion; partial answers are expected during the initial run. +### Weekly conversation export + +Every Monday at 8:00 AM Eastern, an Excel workbook of conversation history is emailed as a download link to whoever is subscribed to the export topic. It is built for product staff who need to read what people asked ABE — and what they rated poorly — without AWS access. + +Name the recipients in `config.yaml` — one address or a list — and have each of them confirm the SNS subscription email AWS sends after the first deploy. Leaving this unset means nobody is emailed; it does not fall back to `notification_email`, so conversation data only reaches addresses named for this export: + +```yaml +export_notification_email: + - pm@example.edu + - programme-lead@example.edu +export_url_expiry_days: 7 # how long the download link stays valid (max 7) +export_retention_days: 90 # optional; unset keeps every export indefinitely +``` + +Each run exports only messages newer than the previous run, tracked by a watermark in SSM Parameter Store (`/abe/conversation-export/last-exported-timestamp`). With no watermark stored, the whole history is exported — so the first email carries a noticeably larger file than later ones. + +The workbook has three sheets: + +| Sheet | Contents | +| --- | --- | +| `conversations` | Every message in this export, one per row, with every stored attribute as its own column. Filters are pre-enabled. | +| `all_feedback` | Every rated message ever, repeated in full on every run. Ratings are written onto the original message row without changing its timestamp, so they fall outside the weekly window and would otherwise be missed. | +| `run_info` | What the export covered, for the record. | + +The email states that the download link expires in 7 days, and carries two plain S3 console links as the fallback — one to that week's file, one to every export kept in the bucket. Exports are kept indefinitely by default, so those links never go stale, but the reader must be signed in to AWS with read access to the export bucket — grant the recipient console access if they will rely on them. + +Note that a presigned URL is signed with the Lambda's temporary credentials, so it can stop working before the stated 7 days if those credentials rotate. The console links are the answer to that; a reliably week-long link would need a dedicated long-lived signing credential or a redirect endpoint in front of the object. + +To run one outside the schedule, invoke the function with an empty payload: + +```bash +aws lambda invoke --function-name abe-conversation-export \ + --cli-binary-format raw-in-base64-out --payload '{}' /dev/stdout +``` + +Send `{"full": true, "advance_watermark": false}` instead to re-export the whole history without disturbing the weekly window. + ## Optional Features These are gated by `config.yaml` flags and are inactive by default. diff --git a/app.py b/app.py index 3b53915..c3916f4 100644 --- a/app.py +++ b/app.py @@ -39,6 +39,17 @@ frontend_certificate_arn=config.get("frontend_certificate_arn"), # Email address for content-sync run notifications (optional) notification_email=config.get("notification_email"), + # Weekly conversation export recipients (optional; one address or a list). + # Deliberately no fallback to notification_email - conversation data goes + # only to addresses named for this export. + export_notification_email=config.get("export_notification_email"), + export_url_expiry_days=int(config.get("export_url_expiry_days", 7)), + # Unset keeps every export indefinitely + export_retention_days=( + int(config["export_retention_days"]) + if config.get("export_retention_days") + else None + ), # Cognito SAML auth — only active when enable_saml_auth: true in config.yaml **({ "cognito_domain_prefix": config.get("cognito_domain_prefix"), diff --git a/cdk/backend.py b/cdk/backend.py index 18ac480..d4895dd 100644 --- a/cdk/backend.py +++ b/cdk/backend.py @@ -87,7 +87,9 @@ def __init__( billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST, removal_policy=RemovalPolicy.DESTROY, ) - + # Read by the weekly conversation export construct + self.conversation_table = conversation_table + ################################################################################# # CDK FOR THE LAMBDA WHICH SERVES THE API ################################################################################# diff --git a/cdk/content_sync.py b/cdk/content_sync.py index a3fb473..aaa04ad 100644 --- a/cdk/content_sync.py +++ b/cdk/content_sync.py @@ -4,6 +4,9 @@ RemovalPolicy, TimeZone, ) +from aws_cdk import ( + aws_dynamodb as dynamodb, +) from aws_cdk import ( aws_ec2 as ec2, ) @@ -67,7 +70,8 @@ def __init__( vpc: ec2.Vpc, input_assets_bucket: s3.Bucket, ingestion_state_machine: sfn.StateMachine, - notification_email: str = None, + processed_files_table: dynamodb.Table, + notification_email: str | list[str] = None, **kwargs, ) -> None: super().__init__(scope, construct_id, **kwargs) @@ -192,10 +196,12 @@ def __init__( # subject line is set per-publish by the notify lambda display_name="ABE content ingestion", ) - if notification_email: - topic.add_subscription( - subscriptions.EmailSubscription(notification_email) - ) + # config.yaml may give one address or a list of them. Each subscription + # has to be confirmed individually from its own inbox. + if isinstance(notification_email, str): + notification_email = [notification_email] + for address in dict.fromkeys(notification_email or []): + topic.add_subscription(subscriptions.EmailSubscription(address)) notify_lambda = lambda_.Function( self, @@ -208,9 +214,11 @@ def __init__( "SNS_TOPIC_ARN": topic.topic_arn, "BUCKET": input_assets_bucket.bucket_name, "COLLECTOR_LOG_GROUP": collector_log_group.log_group_name, + "PROCESSED_FILES_TABLE": processed_files_table.table_name, }, ) topic.grant_publish(notify_lambda) + processed_files_table.grant_read_data(notify_lambda) notify_lambda.add_to_role_policy( iam.PolicyStatement( actions=["s3:GetObject"], diff --git a/cdk/conversation_export.py b/cdk/conversation_export.py new file mode 100644 index 0000000..34d0ff9 --- /dev/null +++ b/cdk/conversation_export.py @@ -0,0 +1,211 @@ +from aws_cdk import ( + BundlingOptions, + CfnOutput, + Duration, + RemovalPolicy, + Stack, + TimeZone, +) +from aws_cdk import ( + aws_dynamodb as dynamodb, +) +from aws_cdk import ( + aws_iam as iam, +) +from aws_cdk import ( + aws_lambda as lambda_, +) +from aws_cdk import ( + aws_s3 as s3, +) +from aws_cdk import ( + aws_scheduler as scheduler, +) +from aws_cdk import ( + aws_scheduler_targets as scheduler_targets, +) +from aws_cdk import ( + aws_sns as sns, +) +from aws_cdk import ( + aws_sns_subscriptions as subscriptions, +) +from constructs import Construct + +WATERMARK_PARAM = "/abe/conversation-export/last-exported-timestamp" + + +class ConversationExport(Construct): + """ + Weekly Excel export of the conversation-history table, emailed as a + presigned download link: + + EventBridge Scheduler (Mondays 8am ET) -> export lambda: + 1. read the watermark from SSM (unset -> export the whole history) + 2. scan the conversation table, build an .xlsx keeping every attribute + 3. upload to the exports bucket, email a presigned link over SNS + 4. advance the watermark + + Aimed at non-engineers: the reader filters the workbook in Excel instead of + querying DynamoDB. The email also carries plain console links to the file, + which keep working after the presigned link expires - those need the reader + to be signed in to AWS with read access to the export bucket. + + The watermark lives outside CDK on purpose - a StringParameter with a value + would reset the export window on every deploy. + """ + + def __init__( + self, + scope: Construct, + construct_id: str, + conversation_table: dynamodb.ITable, + export_email=None, + url_expiry_days: int = 7, + retain_exports_days: int = None, + **kwargs, + ) -> None: + super().__init__(scope, construct_id, **kwargs) + + ################################################################################# + # EXPORT BUCKET + ################################################################################# + # Separate from the content buckets: these files hold what people asked + # ABE, so they expire on their own schedule and are never public. + # + # RETAIN, unlike the rest of the stack: the exports are the historical + # record product staff work from, and old emails link into this bucket, + # so a cdk destroy must not take them with it. The bucket is left + # behind and has to be emptied and deleted by hand if it is ever really + # unwanted. auto_delete_objects is therefore off - CDK rejects it + # without a DESTROY policy. + export_bucket = s3.Bucket( + self, + "ExportBucket", + block_public_access=s3.BlockPublicAccess.BLOCK_ALL, + encryption=s3.BucketEncryption.S3_MANAGED, + enforce_ssl=True, + removal_policy=RemovalPolicy.RETAIN, + # Exports are kept indefinitely unless config.yaml asks for an + # expiry, so the console links in old emails keep working + lifecycle_rules=( + [ + s3.LifecycleRule( + id="expire-old-exports", + prefix="conversation-exports/", + expiration=Duration.days(retain_exports_days), + ) + ] + if retain_exports_days + else [] + ), + ) + + ################################################################################# + # NOTIFICATION TOPIC + ################################################################################# + topic = sns.Topic( + self, + "ConversationExportTopic", + display_name="ABE weekly conversation export", + ) + # config.yaml may give one address or a list of them. Each subscription + # has to be confirmed individually from its own inbox. + if isinstance(export_email, str): + export_email = [export_email] + for address in dict.fromkeys(export_email or []): + topic.add_subscription(subscriptions.EmailSubscription(address)) + + ################################################################################# + # EXPORT LAMBDA + ################################################################################# + export_lambda = lambda_.Function( + self, + "ExportLambda", + # Fixed name so the log group is predictable and scripts/ can invoke + # it without looking up a generated name + function_name="abe-conversation-export", + runtime=lambda_.Runtime.PYTHON_3_13, + handler="export.handler", + code=lambda_.Code.from_asset( + "src/conversation_export", + bundling=BundlingOptions( + image=lambda_.Runtime.PYTHON_3_13.bundling_image, + command=[ + "bash", + "-c", + "pip install --platform manylinux2014_x86_64 --implementation cp --python-version 3.13 --only-binary=:all: --target /asset-output -r requirements.txt && cp -au . /asset-output", + ], + ), + ), + # The whole table is held in memory while the workbook is built; + # generous headroom is cheaper than a failed weekly email. + memory_size=2048, + timeout=Duration.minutes(5), + environment={ + "CONVERSATION_TABLE": conversation_table.table_name, + "EXPORT_BUCKET": export_bucket.bucket_name, + "EXPORT_PREFIX": "conversation-exports", + "SNS_TOPIC_ARN": topic.topic_arn, + "WATERMARK_PARAM": WATERMARK_PARAM, + "URL_EXPIRY_DAYS": str(url_expiry_days), + # 0 tells the email to say the files are kept indefinitely + "EXPORT_RETENTION_DAYS": str(retain_exports_days or 0), + "REPORT_TIMEZONE": "America/New_York", + }, + ) + + conversation_table.grant_read_data(export_lambda) + export_bucket.grant_put(export_lambda, "conversation-exports/*") + # A presigned URL carries the signer's permissions, so the function + # needs read on the object for the emailed link to work at all + export_bucket.grant_read(export_lambda, "conversation-exports/*") + topic.grant_publish(export_lambda) + export_lambda.add_to_role_policy( + iam.PolicyStatement( + actions=["ssm:GetParameter", "ssm:PutParameter"], + resources=[ + f"arn:aws:ssm:{Stack.of(self).region}:" + f"{Stack.of(self).account}:parameter{WATERMARK_PARAM}" + ], + ) + ) + + ################################################################################# + # WEEKLY SCHEDULE + ################################################################################# + # EventBridge Scheduler is timezone-aware, so 8am Eastern stays 8am + # Eastern across DST transitions. + scheduler.Schedule( + self, + "WeeklyExportSchedule", + schedule=scheduler.ScheduleExpression.cron( + minute="0", + hour="8", + week_day="MON", + time_zone=TimeZone.AMERICA_NEW_YORK, + ), + target=scheduler_targets.LambdaInvoke( + export_lambda, input=scheduler.ScheduleTargetInput.from_object({}) + ), + description="Weekly ABE conversation export (Mondays 8am ET)", + ) + + CfnOutput( + self, + "ConversationExportFunctionName", + value=export_lambda.function_name, + description="Invoke with an empty payload to export now instead of waiting for Monday", + ) + CfnOutput( + self, + "ConversationExportBucket", + value=export_bucket.bucket_name, + description="Where weekly conversation exports are stored", + ) + CfnOutput( + self, + "ConversationExportTopicArn", + value=topic.topic_arn, + description="SNS topic for the weekly export email (subscription must be confirmed)", + ) diff --git a/cdk/ingest.py b/cdk/ingest.py index c27e8cd..2cb0fec 100644 --- a/cdk/ingest.py +++ b/cdk/ingest.py @@ -661,6 +661,23 @@ def __init__( security_groups=[security_group], result_path="$.audio_result", ) + process_audio.add_retry( + errors=["States.TaskFailed"], + interval=Duration.seconds(30), + max_attempts=2, + backoff_rate=2.0, + ) + # Infra-level failure (e.g. image pull timeout) after retries are exhausted - + # unlike the Transcribe-failure quarantine branch above, this isn't a bad + # input file, so leave it in place for next run rather than quarantining it. + audio_task_failed = sfn.Succeed( + self, + "AudioTaskFailedNonFatal", + comment="Audio ingestion container failed after retries; file left for next run", + ) + process_audio.add_catch( + audio_task_failed, errors=["States.ALL"], result_path="$.task_error" + ) delete_transcription_job = tasks.CallAwsService( self, @@ -824,6 +841,20 @@ def __init__( security_groups=[security_group], result_path="$.video_result", ) + process_video.add_retry( + errors=["States.TaskFailed"], + interval=Duration.seconds(30), + max_attempts=2, + backoff_rate=2.0, + ) + video_task_failed = sfn.Succeed( + self, + "VideoTaskFailedNonFatal", + comment="Video ingestion container failed after retries; file left for next run", + ) + process_video.add_catch( + video_task_failed, errors=["States.ALL"], result_path="$.task_error" + ) # Separate task for processing video chunks (same task definition, different state) process_video_chunk = tasks.EcsRunTask( @@ -850,6 +881,20 @@ def __init__( security_groups=[security_group], result_path="$.video_result", ) + process_video_chunk.add_retry( + errors=["States.TaskFailed"], + interval=Duration.seconds(30), + max_attempts=2, + backoff_rate=2.0, + ) + video_chunk_task_failed = sfn.Succeed( + self, + "VideoChunkTaskFailedNonFatal", + comment="Video chunk ingestion container failed after retries; chunk left for next run", + ) + process_video_chunk.add_catch( + video_chunk_task_failed, errors=["States.ALL"], result_path="$.task_error" + ) # PDF branch process_pdf = tasks.LambdaInvoke( @@ -1083,10 +1128,12 @@ def __init__( self.processed_files_table_name = processed_files_table.table_name # Exposed for the ContentSync construct, which reuses this - # infrastructure for the collector job + # infrastructure for the collector job and reads ingestion status + # for the notify lambda's per-file report self.cluster = cluster self.vpc = vpc self.input_assets_bucket = input_assets_bucket + self.processed_files_table = processed_files_table self.state_machine = state_machine CfnOutput( diff --git a/cdk/main.py b/cdk/main.py index c73e40c..a53ad75 100644 --- a/cdk/main.py +++ b/cdk/main.py @@ -7,6 +7,7 @@ from .auth import CognitoSamlAuth from .backend import RagBackend from .content_sync import ContentSync +from .conversation_export import ConversationExport from .frontend import RagFrontend from .ingest import RagIngest from .waf import Waf @@ -48,8 +49,13 @@ def __init__( # Custom domain for CloudFront (optional) frontend_domain_name: str = None, frontend_certificate_arn: str = None, - # Email for content-sync run notifications (optional) - notification_email: str = None, + # Email(s) for content-sync run notifications - one address or a list + notification_email: str | list[str] = None, + # Email(s) for the weekly conversation export - one address or a list + export_notification_email: str | list[str] = None, + export_url_expiry_days: int = 7, + # None keeps every export indefinitely + export_retention_days: int = None, **kwargs, ) -> None: super().__init__(scope, construct_id, **kwargs) @@ -135,5 +141,17 @@ def __init__( vpc=ingest_stack.vpc, input_assets_bucket=ingest_stack.input_assets_bucket, ingestion_state_machine=ingest_stack.state_machine, + processed_files_table=ingest_stack.processed_files_table, notification_email=notification_email, ) + + # Weekly Excel export of conversations, emailed as a download link so + # non-engineers can review questions and feedback without AWS access + ConversationExport( + self, + "ConversationExport", + conversation_table=rag_api_stack.conversation_table, + export_email=export_notification_email, + url_expiry_days=export_url_expiry_days, + retain_exports_days=export_retention_days, + ) diff --git a/config.yaml.example b/config.yaml.example index 6c784e3..2484917 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -58,9 +58,33 @@ skip_existing_s3_files: true # Env file with local credentials for manual ingest-script runs (optional, # defaults to names.env in ingest_utils/confluence_processor/) # env_file: names.env -# Email address that receives content-sync run notifications (optional). -# The SNS subscription must be confirmed once via the email AWS sends. +# Who receives content-sync run notifications (optional). Each recipient +# confirms their own SNS subscription once, from the email AWS sends them. +# Keep this authoritative rather than adding people in the SNS console - +# console-added subscribers are invisible to cdk diff. Accepts one address: # notification_email: you@example.edu +# ...or several: +# notification_email: +# - you@example.edu +# - teammate@example.edu + +# Who receives the weekly conversation export - an Excel file of questions, +# answers and ratings, sent Mondays at 8am Eastern as a download link +# (optional). Aimed at product/programme staff who need the data without AWS +# access. Each recipient confirms their own SNS subscription once, from the +# email AWS sends them. Unset means nobody is emailed - this never falls back +# to notification_email, so conversation data only reaches addresses named +# here. Accepts one address: +# export_notification_email: pm@example.edu +# ...or several: +# export_notification_email: +# - pm@example.edu +# - programme-lead@example.edu +# How long the emailed download link stays valid, in days (max 7). +# export_url_expiry_days: 7 +# How long each generated file is kept in S3, in days. Unset keeps every export +# indefinitely, so the console links in old emails never go stale. +# export_retention_days: 90 # Chat file_input_folder: files-to-process/ diff --git a/src/backend/.streamlit/config.toml b/src/backend/.streamlit/config.toml deleted file mode 100644 index d214944..0000000 --- a/src/backend/.streamlit/config.toml +++ /dev/null @@ -1,17 +0,0 @@ -[theme] - -# Primary accent for interactive elements -primaryColor = 'white' - -# Background color for the main content area -backgroundColor = '#343434' - -# Background color for sidebar and most interactive widgets -secondaryBackgroundColor = '#005941' - -# Color used for almost all text -textColor = '#FFFFFF' - - -[browser] -gatherUsageStats = false \ No newline at end of file diff --git a/src/content_sync/notify.py b/src/content_sync/notify.py index 21485da..81b13ca 100644 --- a/src/content_sync/notify.py +++ b/src/content_sync/notify.py @@ -6,6 +6,21 @@ in two tiers: a plain-language summary for non-technical readers, then the raw per-stage counters and AWS console links under a TECHNICAL DETAILS divider. +The reader-facing tier answers exactly one question - of the files this run +tried to add, which made it into ABE and which did not, by name. Routine skips +(file types ABE cannot read, audio dropped in favor of the session video, +Drive folder links handled by the Drive stage) are known and expected, so they +stay in the per-stage counters below the divider instead of being called out +where they read as failures. + +The collector's run_summary only says a file was uploaded to S3 - it says +nothing about whether the separate ingestion state machine then actually +processed it, so a failed ingestion run does not mean every file that run +collected is unsearchable (one branch, e.g. audio, can fail while others +succeed). This lambda checks the ingestion pipeline's own processed-files +table for each uploaded key to report what is actually searchable, rather +than assuming a failed run means nothing made it in. + SNS email is plain text only - no HTML - and subjects must be ASCII and under 100 characters, so all layout here is done with spaces. """ @@ -18,11 +33,15 @@ s3 = boto3.client("s3") sns = boto3.client("sns") +dynamodb = boto3.resource("dynamodb") SNS_TOPIC_ARN = os.environ["SNS_TOPIC_ARN"] BUCKET = os.environ["BUCKET"] COLLECTOR_LOG_GROUP = os.environ["COLLECTOR_LOG_GROUP"] REGION = os.environ["AWS_REGION"] +PROCESSED_FILES_TABLE = os.environ["PROCESSED_FILES_TABLE"] + +processed_files_table = dynamodb.Table(PROCESSED_FILES_TABLE) RULE = "-" * 68 MAX_LISTED_SESSIONS = 60 @@ -247,33 +266,51 @@ def _uploaded_keys(stages): return keys +def _searchable_keys(keys): + """Ground truth for which of this run's uploaded S3 keys are actually + searchable in ABE right now, per the ingestion pipeline's processed-files + table - the only way to tell a file that finished ingestion apart from + one still stuck (or lost) after an ingestion failure. + """ + searchable = set() + for key in keys: + s3_uri = f"s3://{BUCKET}/{key}" + try: + response = processed_files_table.get_item(Key={"s3_uri": s3_uri}) + except Exception as e: + # Can't confirm either way - report it as not-yet-searchable + # rather than crashing the notification + print(f"Could not check processed status for {s3_uri}: {e}") + continue + if "Item" in response: + searchable.add(key) + return searchable + + def _figure(label: str, value, note: str = "") -> str: row = f" {label.ljust(22)}{str(value).rjust(6)}" return f"{row} {note}".rstrip() -def _headline(status: str, failed_stage, added: int, errors: int) -> str: - if status == "SUCCEEDED": - opening = ( - f"Completed with {errors} error{'' if errors == 1 else 's'}" - if errors - else "Completed successfully" - ) - if added: - item = "item" if added == 1 else "items" - return f"{opening} - {added} new {item} added to ABE." - return f"{opening} - no new content found this time." - stage = FAILED_STAGE_LABELS.get(failed_stage, failed_stage or "the pipeline") - return f"FAILED during {stage}." +def _headline(status: str, failed_stage, attempted: int, added: int) -> str: + if status != "SUCCEEDED": + stage = FAILED_STAGE_LABELS.get(failed_stage, failed_stage or "the pipeline") + return f"FAILED during {stage}." + if not attempted: + return "Completed successfully - no new content found this time." + if added == attempted: + item = "file" if added == 1 else "files" + return f"Completed successfully - {added} new {item} added to ABE." + return f"Completed with problems - {added} of {attempted} new files added to ABE." def _what_this_means(failed_stage): lines = ["WHAT THIS MEANS"] if failed_stage == "ingestion": lines += [ - " New files were collected but could not be processed, so they are", - " not searchable in ABE yet. Everything already in ABE is unaffected", - " and the assistant is working normally.", + " Some newly collected files may not be searchable in ABE yet - see", + " the breakdown below for exactly which ones made it in. Everything", + " already in ABE is unaffected and the assistant is working normally.", ] else: lines += [ @@ -290,24 +327,34 @@ def _what_this_means(failed_stage): return lines -def _whats_new_lines(sessions, added: int, status: str): - succeeded = status == "SUCCEEDED" - lines = ["WHAT'S NEW IN ABE" if succeeded else "WHAT WAS COLLECTED"] - if not sessions: - if succeeded: - lines.append( - " Nothing new - every source we check was already up to date in ABE." - ) - else: - lines.append(" Nothing was added to ABE in this run.") - lines.append("") - return lines - - session_word = "session" if len(sessions) == 1 else "sessions" - file_word = "file" if added == 1 else "files" - state = "now searchable in ABE" if succeeded else "not searchable in ABE yet" - lines.append(f" {len(sessions)} {session_word}, {added} {file_word} - {state}:") +def _scoreboard_lines(stages, summary, attempted, added, not_added): + """The one block that answers: how many did we try, and how many made it?""" + lines = ["THIS RUN", _figure("Tried to add", attempted, "(new files found)")] + lines.append(_figure("Added to ABE", added, "(searchable now)" if added else "")) + lines.append( + _figure( + "Not added", + not_added, + "(see breakdown below)" if not_added else "(none)", + ) + ) + lines.append( + _figure( + "Already in ABE", + _total(stages, "already_in_s3"), + "(unchanged since the last run)", + ) + ) + if summary and summary.get("duration_seconds") is not None: + lines.append( + _figure("Collection time", _human_duration(summary["duration_seconds"])) + ) lines.append("") + return lines + + +def _session_lines(sessions): + lines = [] for title, parts in sessions[:MAX_LISTED_SESSIONS]: lines.append(f" - {title} ({', '.join(parts)})") if len(sessions) > MAX_LISTED_SESSIONS: @@ -315,28 +362,37 @@ def _whats_new_lines(sessions, added: int, status: str): f" ... and {len(sessions) - MAX_LISTED_SESSIONS} more " "(full list in the run summary linked below)" ) + return lines + + +def _file_word(count: int) -> str: + return "file" if count == 1 else "files" + + +def _added_lines(sessions, added: int, attempted: int): + lines = [f"ADDED TO ABE - searchable now ({added} {_file_word(added)})"] + if sessions: + lines += _session_lines(sessions) + elif attempted: + lines.append(" - none this run - see below for what happened to the files collected") + else: + lines.append(" - none - every source we check was already up to date") lines.append("") return lines -def _big_picture_lines(stages, added: int, summary, status: str): - already = _total(stages, "already_in_s3") - skipped = _total(stages, "unsupported_skipped", "mp4_dominance_skipped") - errors = _total(stages, "failed") - added_label = ( - "Newly added to ABE" if status == "SUCCEEDED" else "Collected this run" - ) - lines = [ - "THE BIG PICTURE", - _figure(added_label, added), - _figure("Already in ABE", already, "(unchanged since the last run)"), - _figure("Skipped", skipped, "(expected - see below)" if skipped else ""), - _figure("Errors", errors, "" if errors else "(none)"), - ] - if summary and summary.get("duration_seconds") is not None: +def _not_added_lines(sessions, not_added: int, collect_failures: int): + """Name what did not make it, whenever the pipeline can tell us the names.""" + if not not_added: + return [] + lines = [f"NOT ADDED - not searchable yet ({not_added} {_file_word(not_added)})"] + lines += _session_lines(sessions) + if collect_failures: lines.append( - _figure("Collection time", _human_duration(summary["duration_seconds"])) + f" - {collect_failures} {_file_word(collect_failures)} could not be " + "collected from the source" ) + lines.append(" (no file names available - see technical details)") lines.append("") return lines @@ -355,32 +411,6 @@ def _sources_lines(stages): return lines -def _skip_lines(stages): - unsupported = _total(stages, "unsupported_skipped") - dominated = _total(stages, "mp4_dominance_skipped") - deferred = _total(stages, "drive_folders_deferred") - if not (unsupported or dominated or deferred): - return [] - lines = ["WHY SOME ITEMS WERE SKIPPED (all expected)"] - if unsupported: - lines += [ - f" {str(unsupported).rjust(4)} file types ABE cannot read - images, spreadsheets,", - " and links out to other websites", - ] - if dominated: - lines += [ - f" {str(dominated).rjust(4)} audio and caption files skipped because the video of that", - " same session was taken instead - nothing was lost", - ] - if deferred: - lines += [ - f" {str(deferred).rjust(4)} Google Drive folder links - their contents are collected", - " by the Google Drive step instead of one link at a time", - ] - lines.append("") - return lines - - def _technical_lines(summary, run_id, execution_arn, failed_stage, error): lines = [RULE, "TECHNICAL DETAILS", ""] if failed_stage: @@ -433,8 +463,15 @@ def _technical_lines(summary, run_id, execution_arn, failed_stage, error): return lines -def build_email(event, summary): - """Compose the (subject, body) pair. Kept separate from I/O for testing.""" +def build_email(event, summary, searchable_keys=frozenset()): + """Compose the (subject, body) pair. Kept separate from I/O for testing. + + searchable_keys is the ground-truth subset of this run's uploaded S3 keys + that the caller has confirmed are actually processed - see + _searchable_keys(). It is not inferred from the pipeline's status, because + a failed ingestion run does not mean every file that run collected is + unsearchable (one branch, e.g. audio, can fail while others succeed). + """ status = event["status"] # "SUCCEEDED" or "FAILED" run_id = event["run_id"] execution_arn = event["execution_arn"] @@ -442,8 +479,19 @@ def build_email(event, summary): error = event.get("error") stages = (summary or {}).get("stages") or {} - sessions = _group_sessions(_uploaded_keys(stages)) - added = _total(stages, "uploaded") + uploaded_keys = _uploaded_keys(stages) + added_keys = [k for k in uploaded_keys if k in searchable_keys] + not_added_keys = [k for k in uploaded_keys if k not in searchable_keys] + added_sessions = _group_sessions(added_keys) + not_added_sessions = _group_sessions(not_added_keys) + + # Everything the run tried to put into ABE: files it collected, plus files it + # tried to collect and could not. + collected = _total(stages, "uploaded") + collect_failures = _total(stages, "failed") + attempted = collected + collect_failures + added = len(added_keys) + not_added = len(not_added_keys) + collect_failures run_dt = _run_datetime(run_id) when = ( @@ -456,26 +504,30 @@ def build_email(event, summary): "ABE CONTENT INGESTION", when, "", - _headline(status, failed_stage, added, _total(stages, "failed")), + _headline(status, failed_stage, attempted, added), "", ] if status != "SUCCEEDED": lines += _what_this_means(failed_stage) - lines += _whats_new_lines(sessions, added, status) if stages: - lines += _big_picture_lines(stages, added, summary, status) + lines += _scoreboard_lines(stages, summary, attempted, added, not_added) + lines += _added_lines(added_sessions, added, attempted) + lines += _not_added_lines(not_added_sessions, not_added, collect_failures) lines += _sources_lines(stages) - lines += _skip_lines(stages) + else: + lines += ["Nothing was added to ABE in this run.", ""] lines += _technical_lines(summary, run_id, execution_arn, failed_stage, error) if status == "SUCCEEDED": - if added: - item = "item" if added == 1 else "items" + if not attempted: + subject = "ABE content ingestion: no new content" + elif added == attempted: + item = "file" if added == 1 else "files" subject = f"ABE content ingestion: {added} new {item} added" else: - subject = "ABE content ingestion: no new content" - elif failed_stage == "ingestion": - subject = "ABE content ingestion FAILED - new content not searchable yet" + subject = f"ABE content ingestion: {added} of {attempted} new files added" + elif attempted: + subject = f"ABE content ingestion FAILED - {added} of {attempted} new files added" else: subject = "ABE content ingestion FAILED - no new content added" if run_dt: @@ -488,6 +540,8 @@ def build_email(event, summary): def handler(event, context): summary = _load_run_summary(event["run_id"]) - subject, body = build_email(event, summary) + stages = (summary or {}).get("stages") or {} + searchable_keys = _searchable_keys(_uploaded_keys(stages)) + subject, body = build_email(event, summary, searchable_keys) sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=subject, Message=body) return {"published": True, "status": event["status"]} diff --git a/src/conversation_export/export.py b/src/conversation_export/export.py new file mode 100644 index 0000000..6c8bd2e --- /dev/null +++ b/src/conversation_export/export.py @@ -0,0 +1,616 @@ +""" +Weekly conversation export for ABE. + +Runs Mondays at 8:00 AM Eastern (EventBridge Scheduler). Each run: + + 1. reads the watermark - the newest message timestamp covered by the previous + export - from SSM Parameter Store + 2. scans the conversation-history table and keeps everything newer than the + watermark. With no watermark stored the whole history is exported, so the + first email carries a complete (larger) workbook and later ones only the + new week. + 3. builds an .xlsx that keeps every DynamoDB attribute as its own column, so + the reader filters in Excel instead of asking for a different export + 4. uploads it and emails a presigned download link over SNS + 5. advances the watermark only after SNS accepts the message, so a failed + send is picked up by the next run instead of silently losing a week + +Two details worth knowing before changing this: + +* Feedback is written onto the original message row (thumb_rating, + feedback_text) without changing its timestamp, so a timestamp window alone + would miss a thumbs-down left on an older answer. Every rated message + therefore ships in its own sheet on every run. +* A DynamoDB FilterExpression is applied after the rows are read, so filtering + the scan server-side would cost exactly the same as reading the table and + splitting the rows here. One full scan feeds both sheets. +""" +import io +import json +import os +import re +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from urllib.parse import quote +from zoneinfo import ZoneInfo + +import boto3 +from botocore.exceptions import ClientError +from openpyxl import Workbook +from openpyxl.styles import Font +from openpyxl.utils import get_column_letter + +TABLE_NAME = os.environ.get("CONVERSATION_TABLE", "") +EXPORT_BUCKET = os.environ.get("EXPORT_BUCKET", "") +EXPORT_PREFIX = os.environ.get("EXPORT_PREFIX", "conversation-exports") +SNS_TOPIC_ARN = os.environ.get("SNS_TOPIC_ARN", "") +WATERMARK_PARAM = os.environ.get("WATERMARK_PARAM", "") +URL_EXPIRY_DAYS = int(os.environ.get("URL_EXPIRY_DAYS", "7")) +# 0 or unset means exports are kept indefinitely +RETENTION_DAYS = int(os.environ.get("EXPORT_RETENTION_DAYS", "0")) +REGION = os.environ.get("AWS_REGION", "us-east-1") +REPORT_TIMEZONE = os.environ.get("REPORT_TIMEZONE", "America/New_York") + +dynamodb = boto3.resource("dynamodb") +s3 = boto3.client("s3") +sns = boto3.client("sns") +ssm = boto3.client("ssm") + +RULE = "-" * 68 +NEXT_RUN = "Mondays at 8:00 AM Eastern" +XLSX_CONTENT_TYPE = ( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" +) + +# Excel refuses these control characters and truncates a cell past 32767 chars +_ILLEGAL_CELL_CHARS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]") +MAX_CELL_CHARS = 32767 + +# Written on every run even when this week's rows never use them, so the column +# layout - and therefore any Excel filter or pivot built on it - stays put. +BASE_COLUMNS = ( + "session_id", + "timestamp", + "role", + "content", + "conversation_turn", + "owner_sub", + "thumb_rating", + "feedback_text", + "document_ids", + "sources", +) + +# Added by the export for readability; the raw attributes above are all kept +DERIVED_COLUMNS = ( + "datetime_eastern", + "date_eastern", + "datetime_utc", + "source_count", + "sources_titles", +) + +COLUMN_WIDTHS = { + "datetime_eastern": 20, + "date_eastern": 13, + "datetime_utc": 22, + "source_count": 8, + "sources_titles": 60, + "session_id": 38, + "timestamp": 16, + "role": 11, + "content": 90, + "conversation_turn": 38, + "owner_sub": 38, + "thumb_rating": 14, + "feedback_text": 24, + "document_ids": 40, + "sources": 60, +} + +# Excel shows a datetime as a raw serial number without one of these +NUMBER_FORMATS = { + "datetime_eastern": "yyyy-mm-dd hh:mm", + "date_eastern": "yyyy-mm-dd", +} + +CONVERSATIONS_SHEET = "conversations" +FEEDBACK_SHEET = "all_feedback" +RUN_INFO_SHEET = "run_info" + + +def _report_tz(): + try: + return ZoneInfo(REPORT_TIMEZONE) + except Exception as e: + # tzdata is a dependency, but a missing zone must not lose the export + print(f"Could not load timezone {REPORT_TIMEZONE}, using UTC: {e}") + return timezone.utc + + +REPORT_TZ = _report_tz() + + +##################################################################### +# VALUE CONVERSION +##################################################################### +def _plain(value): + """Make a DynamoDB-decoded value JSON-serializable.""" + if isinstance(value, Decimal): + number = float(value) + return int(number) if number.is_integer() else number + if isinstance(value, dict): + return {key: _plain(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_plain(item) for item in value] + if isinstance(value, set): + return sorted(_plain(item) for item in value) + if isinstance(value, (bytes, bytearray)): + return value.decode("utf-8", errors="replace") + return value + + +def _cell(value): + """Render one attribute as something Excel will accept.""" + if value is None or isinstance(value, bool): + return value + if isinstance(value, Decimal): + return _plain(value) + if isinstance(value, (int, float, datetime)): + return value + if isinstance(value, (list, tuple, dict, set)): + value = json.dumps(_plain(value), ensure_ascii=False, default=str) + text = _ILLEGAL_CELL_CHARS.sub("", str(value)) + if len(text) > MAX_CELL_CHARS: + text = text[: MAX_CELL_CHARS - 15] + " ...[truncated]" + return text + + +def _as_int(value, default=0): + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _eastern(timestamp_ms): + """Epoch milliseconds -> naive local datetime (Excel rejects tz-aware).""" + if timestamp_ms is None: + return None + moment = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc) + return moment.astimezone(REPORT_TZ).replace(tzinfo=None) + + +def _utc_iso(timestamp_ms): + if timestamp_ms is None: + return None + moment = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc) + return moment.strftime("%Y-%m-%d %H:%M:%S UTC") + + +def _sources_titles(item): + sources = item.get("sources") + if not isinstance(sources, list): + return None + titles = [ + str(source.get("title")).strip() + for source in sources + if isinstance(source, dict) and source.get("title") + ] + # A single answer often cites the same deck several times + return " | ".join(dict.fromkeys(titles)) or None + + +def _derived(item): + timestamp_ms = _as_int(item.get("timestamp"), default=None) + local = _eastern(timestamp_ms) + sources = item.get("sources") + return { + "datetime_eastern": local, + "date_eastern": local.date() if local else None, + "datetime_utc": _utc_iso(timestamp_ms), + "source_count": len(sources) if isinstance(sources, list) else None, + "sources_titles": _sources_titles(item), + } + + +##################################################################### +# WORKBOOK +##################################################################### +def _columns_for(items): + """Stable base layout first, then any attribute added to the table later.""" + seen = set() + for item in items: + seen.update(item.keys()) + extras = sorted(seen - set(BASE_COLUMNS)) + return list(DERIVED_COLUMNS) + list(BASE_COLUMNS) + extras + + +def _sorted_for_reading(items): + """ + Group each conversation's messages together, conversations oldest first. + + Sorting on timestamp alone interleaves concurrent sessions, which makes a + question and its answer hard to read side by side. + """ + session_start = {} + for item in items: + session = str(item.get("session_id")) + timestamp = _as_int(item.get("timestamp")) + if session not in session_start or timestamp < session_start[session]: + session_start[session] = timestamp + return sorted( + items, + key=lambda item: ( + session_start.get(str(item.get("session_id")), 0), + str(item.get("session_id")), + _as_int(item.get("timestamp")), + ), + ) + + +def _apply_column_styles(sheet, columns): + for index, column in enumerate(columns, start=1): + letter = get_column_letter(index) + sheet.column_dimensions[letter].width = COLUMN_WIDTHS.get(column, 20) + number_format = NUMBER_FORMATS.get(column) + if number_format: + for cell in sheet[letter][1:]: + cell.number_format = number_format + + +def _write_records_sheet(workbook, title, items, columns): + sheet = workbook.create_sheet(title) + sheet.append(columns) + for cell in sheet[1]: + cell.font = Font(bold=True) + + for item in _sorted_for_reading(items): + row = dict(_derived(item)) + for column in columns: + if column not in DERIVED_COLUMNS: + row[column] = item.get(column) + sheet.append([_cell(row.get(column)) for column in columns]) + + _apply_column_styles(sheet, columns) + + # Ready to filter the moment the file opens - the whole point of the export + sheet.freeze_panes = "A2" + if sheet.max_row >= 1: + last = get_column_letter(len(columns)) + sheet.auto_filter.ref = f"A1:{last}{max(sheet.max_row, 1)}" + return sheet + + +def _write_run_info_sheet(workbook, meta): + sheet = workbook.create_sheet(RUN_INFO_SHEET) + sheet.append(["field", "value"]) + for cell in sheet[1]: + cell.font = Font(bold=True) + for field, value in meta.items(): + sheet.append([field, _cell(value)]) + sheet.column_dimensions["A"].width = 30 + sheet.column_dimensions["B"].width = 70 + return sheet + + +def build_workbook(new_items, rated_items, meta): + """Compose the workbook. Kept free of I/O so it can be tested directly.""" + workbook = Workbook() + workbook.remove(workbook.active) + columns = _columns_for(list(new_items) + list(rated_items)) + _write_records_sheet(workbook, CONVERSATIONS_SHEET, new_items, columns) + _write_records_sheet(workbook, FEEDBACK_SHEET, rated_items, columns) + _write_run_info_sheet(workbook, meta) + buffer = io.BytesIO() + workbook.save(buffer) + return buffer.getvalue() + + +##################################################################### +# DATA +##################################################################### +def scan_all_messages(): + table = dynamodb.Table(TABLE_NAME) + items = [] + kwargs = {} + while True: + page = table.scan(**kwargs) + items += page.get("Items", []) + last_key = page.get("LastEvaluatedKey") + if not last_key: + return items + kwargs["ExclusiveStartKey"] = last_key + + +def load_watermark(): + if not WATERMARK_PARAM: + return None + try: + value = ssm.get_parameter(Name=WATERMARK_PARAM)["Parameter"]["Value"] + return int(value) + except ssm.exceptions.ParameterNotFound: + # First ever run: no watermark means "export everything" + return None + except (ClientError, TypeError, ValueError) as e: + # A corrupt watermark would silently truncate the export, so fail loudly + raise RuntimeError(f"Could not read watermark {WATERMARK_PARAM}: {e}") from e + + +def save_watermark(timestamp_ms): + ssm.put_parameter( + Name=WATERMARK_PARAM, + Value=str(int(timestamp_ms)), + Type="String", + Overwrite=True, + Description="Newest conversation timestamp covered by a weekly export", + ) + + +##################################################################### +# EMAIL +##################################################################### +def _figure(label, value, note=""): + row = f" {label.ljust(24)}{str(value).rjust(6)}" + return f"{row} {note}".rstrip() + + +def _window_phrase(watermark_ms, mode): + if mode == "full": + return "the complete history (first export)" + start = _eastern(watermark_ms) + if not start: + return "the complete history" + return f"messages after {start:%b} {start.day}, {start.year} {start:%H:%M} Eastern" + + +def _retention_lines(): + if RETENTION_DAYS: + return [ + " Save the file somewhere of your own if you want to keep it -", + f" copies in AWS are deleted after {RETENTION_DAYS} days.", + ] + return [ + " Every past export is kept indefinitely, so the console links above", + " keep working however long ago the email was sent.", + ] + + +def _console_object_url(key): + """Console link to this one file - works after the presigned link expires.""" + return ( + f"https://{REGION}.console.aws.amazon.com/s3/object/{EXPORT_BUCKET}" + f"?region={REGION}&bucketType=general&prefix={quote(key)}" + ) + + +def _console_prefix_url(): + """Console link to every export kept in the bucket.""" + return ( + f"https://{REGION}.console.aws.amazon.com/s3/buckets/{EXPORT_BUCKET}" + f"?region={REGION}&bucketType=general&prefix={quote(EXPORT_PREFIX)}/" + ) + + +def build_email(stats, meta, url, filename): + """Compose the (subject, body) pair. Kept separate from I/O for testing.""" + generated = stats["generated_eastern"] + new_rows = stats["new_rows"] + questions = stats["questions"] + answers = stats["answers"] + sessions = stats["sessions"] + expires = stats["expires_eastern"] + + if new_rows: + conversation_word = "conversation" if sessions == 1 else "conversations" + headline = ( + f"{new_rows} new messages across {sessions} {conversation_word} " + "- the Excel file is ready to download." + ) + else: + headline = ( + "No new questions were asked this week. The file still has every " + "rating ever left, in case you want to review those." + ) + + lines = [ + "ABE CONVERSATION EXPORT", + f"{generated:%A, %B} {generated.day}, {generated.year}", + "", + headline, + "", + "DOWNLOAD", + f" {url}", + "", + f" THIS LINK EXPIRES IN {URL_EXPIRY_DAYS} DAYS, on " + f"{expires:%A, %B} {expires.day}, {expires.year}.", + f" Covers {_window_phrase(stats['watermark_ms'], stats['mode'])}.", + "", + "AFTER THE LINK EXPIRES", + " The file itself is not going anywhere. Sign in to the AWS console", + " first, then open it directly:", + "", + f" {_console_object_url(filename)}", + "", + " Every past export is listed here:", + "", + f" {_console_prefix_url()}", + "", + " Both links need you to be signed in to AWS - open them in the same", + " browser where you are signed in, then use the Download button.", + "", + "THE BIG PICTURE", + _figure("Questions asked", questions), + _figure("Answers returned", answers), + _figure("Conversations", sessions), + _figure("Thumbs up (this batch)", stats["thumbs_up_new"]), + _figure("Thumbs down (this batch)", stats["thumbs_down_new"]), + _figure("Rated messages, all time", stats["rated_total"], "(sheet 2)"), + "", + "WHAT'S IN THE FILE", + f" Sheet 1 {CONVERSATIONS_SHEET.ljust(15)}" + "every message in this export, one per row", + f" Sheet 2 {FEEDBACK_SHEET.ljust(15)}" + "every thumbs up/down ever left, with its reason", + f" Sheet 3 {RUN_INFO_SHEET.ljust(15)}" + "what this export covered, for the record", + "", + "HOW TO USE IT", + " Every field stored per message is its own column and filters are", + " already switched on, so you should not need a different export:", + "", + " - filter role = user to read just the questions people asked", + " - filter thumb_rating = thumbs_down to see what fell short, with", + " the reason in feedback_text", + " - sources_titles lists the documents ABE cited in each answer", + " - a question and its answer share a conversation_turn value and", + " sit next to each other", + "", + " Sheet 2 exists because a rating can be left days after the answer.", + " Those rows would fall outside this week's window, so every rated", + " message is repeated there in full.", + "", + *_retention_lines(), + "", + f" The next export is sent {NEXT_RUN}.", + "", + RULE, + "TECHNICAL DETAILS", + "", + f"File: s3://{EXPORT_BUCKET}/{filename}", + f"Table: {TABLE_NAME}", + f"Mode: {stats['mode']}", + f"Watermark before this run: {stats['watermark_ms'] or 'unset (full export)'}", + f"Watermark after this run: {stats['new_watermark_ms'] or 'unchanged'}", + f"Rows scanned: {stats['scanned_rows']}", + f"Rows written to sheet 1: {new_rows}", + f"Rows written to sheet 2: {stats['rated_total']}", + f"Presigned link requested for: {URL_EXPIRY_DAYS} days", + "", + "The presigned link is signed with this function's temporary", + "credentials, so it can stop working before the stated expiry if those", + "credentials rotate. The console links above are the fallback and need", + "the reader to have console access to the export bucket.", + ] + for field, value in meta.items(): + if field.startswith("export_"): + lines.append(f"{field}: {value}") + + subject = ( + f"ABE conversations: {new_rows} new messages" + if new_rows + else "ABE conversations: no new activity" + ) + subject = f"{subject} - {generated:%b} {generated.day}" + # SNS subjects must be ASCII and under 100 chars + subject = subject.encode("ascii", errors="ignore").decode().strip()[:100] + return subject, "\n".join(lines) + + +##################################################################### +# S3 +##################################################################### +def _presign(key, filename): + return s3.generate_presigned_url( + "get_object", + Params={ + "Bucket": EXPORT_BUCKET, + "Key": key, + # Without this the browser saves the file under its full key path + "ResponseContentDisposition": ( + f'attachment; filename="{os.path.basename(filename)}"' + ), + "ResponseContentType": XLSX_CONTENT_TYPE, + }, + ExpiresIn=URL_EXPIRY_DAYS * 86400, + ) + + +##################################################################### +# HANDLER +##################################################################### +def handler(event, context): + event = event or {} + full = bool(event.get("full")) + watermark_ms = None if full else load_watermark() + if event.get("since_ms") is not None: + watermark_ms = int(event["since_ms"]) + mode = "full" if watermark_ms is None else "incremental" + + all_items = scan_all_messages() + new_items = [ + item + for item in all_items + if watermark_ms is None or _as_int(item.get("timestamp")) > watermark_ms + ] + rated_items = [item for item in all_items if item.get("thumb_rating")] + + generated = datetime.now(timezone.utc) + generated_local = generated.astimezone(REPORT_TZ).replace(tzinfo=None) + timestamps = [_as_int(item.get("timestamp")) for item in new_items] + new_watermark_ms = max(timestamps) if timestamps else watermark_ms + + stats = { + "mode": mode, + "generated_eastern": generated_local, + "expires_eastern": generated_local + timedelta(days=URL_EXPIRY_DAYS), + "watermark_ms": watermark_ms, + "new_watermark_ms": new_watermark_ms if timestamps else None, + "scanned_rows": len(all_items), + "new_rows": len(new_items), + "questions": sum(1 for item in new_items if item.get("role") == "user"), + "answers": sum(1 for item in new_items if item.get("role") == "assistant"), + "sessions": len({str(item.get("session_id")) for item in new_items}), + "thumbs_up_new": sum( + 1 for item in new_items if item.get("thumb_rating") == "thumbs_up" + ), + "thumbs_down_new": sum( + 1 for item in new_items if item.get("thumb_rating") == "thumbs_down" + ), + "rated_total": len(rated_items), + } + + meta = { + "export_generated_eastern": f"{generated_local:%Y-%m-%d %H:%M}", + "export_generated_utc": generated.strftime("%Y-%m-%d %H:%M:%S UTC"), + "export_mode": mode, + "export_covers": _window_phrase(watermark_ms, mode), + "export_rows_sheet1": stats["new_rows"], + "export_rows_sheet2": stats["rated_total"], + "export_questions": stats["questions"], + "export_answers": stats["answers"], + "export_conversations": stats["sessions"], + "export_table": TABLE_NAME, + "export_watermark_before": watermark_ms or "unset (full export)", + "export_watermark_after": new_watermark_ms or "unchanged", + } + + workbook = build_workbook(new_items, rated_items, meta) + stamp = generated.strftime("%Y%m%d-%H%M") + key = f"{EXPORT_PREFIX}/abe-conversations-{stamp}.xlsx" + s3.put_object( + Bucket=EXPORT_BUCKET, + Key=key, + Body=workbook, + ContentType=XLSX_CONTENT_TYPE, + ) + print(f"Wrote {len(workbook)} bytes to s3://{EXPORT_BUCKET}/{key}") + + url = _presign(key, key) + subject, body = build_email(stats, meta, url, key) + sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=subject, Message=body) + + # Only now is the week safely delivered; a send failure above leaves the + # watermark alone so the next run re-exports this window. + if timestamps and event.get("advance_watermark", True): + save_watermark(new_watermark_ms) + print(f"Watermark advanced to {new_watermark_ms}") + + return { + "key": key, + "bytes": len(workbook), + "new_rows": stats["new_rows"], + "rated_total": stats["rated_total"], + "mode": mode, + "watermark": new_watermark_ms, + } diff --git a/src/conversation_export/requirements.txt b/src/conversation_export/requirements.txt new file mode 100644 index 0000000..700e051 --- /dev/null +++ b/src/conversation_export/requirements.txt @@ -0,0 +1,3 @@ +openpyxl==3.1.5 +# zoneinfo needs an IANA database, which the Lambda image does not ship +tzdata==2025.2 diff --git a/src/ingest/audio/main.py b/src/ingest/audio/main.py index 3ed34ff..cdfd7ce 100644 --- a/src/ingest/audio/main.py +++ b/src/ingest/audio/main.py @@ -145,8 +145,9 @@ def main(transcript_uri, media_file_uri, job_name, metadata): except Exception as e: print(f"Error processing step function input: {e}") + sys.exit(1) result = main(transcript_uri, media_file_uri, job_name, metadata) print(result) - sys.exit(0) + sys.exit(0 if result.get("statusCode") == 200 else 1) diff --git a/src/ingest/video/main.py b/src/ingest/video/main.py index 8702390..d00fd7d 100644 --- a/src/ingest/video/main.py +++ b/src/ingest/video/main.py @@ -268,8 +268,9 @@ def main(media_file_uri, transcribe_uri, metadata): except Exception as e: logger.error(f"Error processing step function input: {e}") + sys.exit(1) result = main(media_file_uri, transcript_uri, metadata) print(result) - sys.exit(0) + sys.exit(0 if result else 1)