feat: Automate YouTube ingestion and auto-commenting via background scheduler - #63
Open
parthdude07 wants to merge 4 commits into
Open
feat: Automate YouTube ingestion and auto-commenting via background scheduler#63parthdude07 wants to merge 4 commits into
parthdude07 wants to merge 4 commits into
Conversation
📝 WalkthroughWalkthroughChangesYouTube commenting and ingestion automation
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Scheduler
participant IngestionService
participant ChannelScanner
participant TranscriptionExport
participant YouTubeCommenterService
participant YouTubeAPI
Scheduler->>IngestionService: run_full_pipeline()
IngestionService->>ChannelScanner: scan channels and queue videos
IngestionService->>TranscriptionExport: process queued transcript
TranscriptionExport->>YouTubeCommenterService: comment_from_transcript_object()
YouTubeCommenterService->>YouTubeAPI: insert transcript summary comment
YouTubeAPI-->>YouTubeCommenterService: posted comment metadata
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
parthdude07
force-pushed
the
yt-commenter
branch
from
July 25, 2026 09:46
f248370 to
2236181
Compare
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Review completed against the latest diff
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 15
🧹 Nitpick comments (3)
app/transcription.py (1)
608-611: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winPreserve the traceback for nonfatal auto-comment failures.
The broad catch is intentional for export resilience, but interpolating only
{e}drops the traceback. Uselogger.exception()and narrow the exception types where practical.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/transcription.py` around lines 608 - 611, Update the auto-comment failure handler in the surrounding transcription flow to use self.logger.exception instead of warning so nonfatal failures retain their traceback, while preserving the existing resilience behavior. Narrow the broad Exception catch to the specific expected exception types where practical, without changing successful export behavior.Source: Linters/SAST tools
app/scheduler.py (1)
17-21: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winKeep scheduler failures observable.
The broad exception handler suppresses all pipeline failures and logs only the exception text. Use traceback logging plus a metric/alert so a permanently failing scheduler is not mistaken for a healthy one.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/scheduler.py` around lines 17 - 21, Update the exception handling around IngestionService.run_full_pipeline to log the full traceback rather than only the exception text, and emit the project’s established failure metric or alert for scheduler pipeline failures. Preserve the existing scheduled execution while ensuring repeated permanent failures remain observable.Source: Linters/SAST tools
app/services/yt_commenter.py (1)
53-62: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate-check logic repeated across three methods.
comment_on_video,comment_with_summary, andcomment_from_transcript_objecteach re-implement the same "look up existing comment, returnalready_commented" block. Extracting a shared helper would reduce drift risk if the dedup behavior needs to change.def _check_existing(self, video_id: str) -> Optional[dict]: existing = self._db.get_yt_comment_by_video_id(video_id) if existing: logger.info(f"Already commented on video {video_id}, skipping.") return { "status": "already_commented", "video_id": video_id, "comment_id": existing["comment_id"], "posted_at": existing.get("posted_at"), } return NoneAlso applies to: 104-113, 141-149
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/services/yt_commenter.py` around lines 53 - 62, Extract the repeated existing-comment lookup and “already_commented” response into a shared _check_existing(video_id) helper on the commenter class. Update comment_on_video, comment_with_summary, and comment_from_transcript_object to call this helper and return its result when present, preserving the existing logging and response fields while using a safe optional posted_at lookup.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/scheduler.py`:
- Around line 24-43: Make ingestion scheduling globally singleton-owned: update
app/scheduler.py’s start_scheduler to acquire an external scheduler or
distributed lock before creating or starting the BackgroundScheduler, and
release ownership during shutdown. Update server.py’s FastAPI lifespan startup
so workers only start the scheduler when elected and globally gated; apply the
coordination changes at both listed sites.
- Around line 47-52: Update stop_scheduler() to wait for running scheduler jobs
to finish before returning by using blocking shutdown behavior. Ensure the
existing _run_ingestion_job() execution completes before FastAPI teardown
proceeds, while preserving the current scheduler-running guard and log message.
In `@app/services/channel_scanner.py`:
- Around line 119-122: Update the channel scanning flow around last_scanned and
update_channel_scanned() to preserve and reuse the persisted scan cursor instead
of resetting it to None. Ensure each run pages through the backlog or otherwise
continues from the saved boundary until the pending older videos are safely
considered, while retaining the existing limit of selecting up to 7 new videos
before marking the scan complete.
In `@app/services/database_service.py`:
- Around line 499-512: Resolve the undefined YouTubeComment used by
save_yt_comment: either add and import a model whose fields match comment_data,
or replace the model construction with an existing database model and explicit
field mapping. Ensure the inserted fields align with the schema and that
save_yt_comment can instantiate and persist the record without unresolved
symbols or unexpected arguments.
- Around line 571-589: Update get_transcript_by_video_id so video_id is escaped
before constructing the ILIKE pattern, treating underscores and percent signs as
literal characters. Pass the matching escape character through the ILIKE
expression’s escape option, preserving the existing availability check,
first-match behavior, and error handling.
In `@app/services/ingestion_service.py`:
- Around line 121-126: Update IngestionService._submit_to_pipeline() to pass an
explicit finite timeout to the requests.post() call targeting
/transcription/add_to_queue/. Preserve the existing submission behavior while
ensuring a stalled queue request raises or returns through the existing
error-handling path instead of blocking indefinitely.
In `@app/services/youtube_auth.py`:
- Around line 25-82: Update get_authenticated_youtube_service so that when no
valid credentials remain, it does not invoke InstalledAppFlow.run_local_server;
detect the non-interactive/headless runtime and raise the existing
setup-directed RuntimeError immediately instead. Preserve token loading,
refresh, and interactive authorization only for supported interactive contexts,
ensuring scheduler and request threads fail fast.
In `@app/services/yt_commenter.py`:
- Around line 353-354: Update the exception handler surrounding video ID
extraction in the relevant method of YTCommenter to stop silently swallowing
parsing failures. Replace the bare pass with the existing logging mechanism,
recording the exception and malformed-URL context while preserving the current
fallback behavior.
- Around line 216-273: Update _post_comment so it validates the result of
self._db.save_yt_comment(comment_data) for the successful YouTube post and
treats a falsey/failed result as a persistence error, rather than returning
"posted". Ensure this failure is surfaced to the caller so retry logic cannot
silently proceed without a deduplication record; preserve the existing
failed-attempt handling for HttpError.
In `@app/transcription.py`:
- Around line 604-607: Update the logging in comment_from_transcript_object() to
inspect the returned status instead of treating every truthy result as a
successful post. Log posted outcomes as auto-commented, duplicate results with
status="already_commented" as skipped, and failed or other non-success outcomes
using an appropriate failure message.
In `@docs/automated_ingestion_updates.md`:
- Around line 7-9: Update the ingestion documentation to describe the
implemented batch behavior: ChannelScanner fetches up to 20 results and
processes up to 7 new video IDs, rather than exactly one latest video. Apply
this correction in docs/automated_ingestion_updates.md lines 7-9 and
docs/competency_test_proof.md lines 27-33, preserving the remaining documented
behavior.
In `@docs/competency_test_proof.md`:
- Around line 43-54: Replace the “Insert Screenshot Here” placeholders in the
“Proof of Execution” section with actual execution screenshots and supporting
evidence, or explicitly mark the document as an unverified template instead of
presenting placeholders as proof.
In `@docs/local-run-and-queue.md`:
- Line 12: Update the local run instructions to remove hard-coded
virtual-environment paths and use the portable ingestion queue request,
including the correct /ingestion/run endpoint. Replace the repeated Uvicorn
startup step with a curl POST command, and document that the request queues the
expected 10-video flow.
In `@routes/yt_commenter.py`:
- Around line 50-54: Update the generic 500-error handlers in the affected
request flows, including the handler around the shown post-comment logic, to
keep logging the original exception while replacing HTTPException.detail=str(e)
with a generic client-safe internal-server-error message. Preserve the existing
404 ValueError handling and status codes.
In `@scraper`:
- Line 1: Add a .gitmodules entry mapping the scraper gitlink to its correct
repository and pinned revision, then update the project’s checkout or CI
initialization flow to fetch submodules before any scraper code is used. Ensure
fresh checkouts have scraper available at runtime without altering unrelated
components.
---
Nitpick comments:
In `@app/scheduler.py`:
- Around line 17-21: Update the exception handling around
IngestionService.run_full_pipeline to log the full traceback rather than only
the exception text, and emit the project’s established failure metric or alert
for scheduler pipeline failures. Preserve the existing scheduled execution while
ensuring repeated permanent failures remain observable.
In `@app/services/yt_commenter.py`:
- Around line 53-62: Extract the repeated existing-comment lookup and
“already_commented” response into a shared _check_existing(video_id) helper on
the commenter class. Update comment_on_video, comment_with_summary, and
comment_from_transcript_object to call this helper and return its result when
present, preserving the existing logging and response fields while using a safe
optional posted_at lookup.
In `@app/transcription.py`:
- Around line 608-611: Update the auto-comment failure handler in the
surrounding transcription flow to use self.logger.exception instead of warning
so nonfatal failures retain their traceback, while preserving the existing
resilience behavior. Narrow the broad Exception catch to the specific expected
exception types where practical, without changing successful export behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 43536349-27b5-455f-9e93-0c2d3829661a
📒 Files selected for processing (21)
.gitignoreREADME.mdapp/config.pyapp/scheduler.pyapp/services/channel_scanner.pyapp/services/database_service.pyapp/services/ingestion_service.pyapp/services/youtube_auth.pyapp/services/yt_commenter.pyapp/transcription.pydocs/automated_ingestion_updates.mddocs/competency_test_proof.mddocs/local-run-and-queue.mdenv.examplerequirements.txtroutes/ingestion.pyroutes/transcription.pyroutes/yt_commenter.pyscraperserver.pytest/test_yt_commenter.py
Comment on lines
+24
to
+43
| def start_scheduler(): | ||
| """Start the background scheduler for periodic tasks.""" | ||
| global _scheduler | ||
| if _scheduler is not None and _scheduler.running: | ||
| logger.warning("Scheduler is already running.") | ||
| return | ||
|
|
||
| logger.info("Starting background scheduler...") | ||
| _scheduler = BackgroundScheduler() | ||
|
|
||
| # Schedule the ingestion pipeline to run every 5 minutes | ||
| _scheduler.add_job( | ||
| _run_ingestion_job, | ||
| trigger=IntervalTrigger(minutes=5), | ||
| id="ingestion_pipeline", | ||
| name="Run ingestion pipeline periodically", | ||
| replace_existing=True, | ||
| ) | ||
|
|
||
| _scheduler.start() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | sed -n '1,200p'
echo "== scheduler outline =="
if [ -f app/scheduler.py ]; then
ast-grep outline app/scheduler.py || true
wc -l app/scheduler.py
cat -n app/scheduler.py
fi
echo "== server references =="
rg -n "lifespan|start_scheduler|stop_scheduler|uvicorn|server|scheduler" -S .
echo "== changed files stat =="
git diff --stat HEAD~1..HEAD 2>/dev/null || true
git diff --stat 2>/dev/null || trueRepository: genesis-kb/transcription_engine
Length of output: 22239
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
# Read-only behavioral probe for the deterministic FastAPI/uvicorn startup pattern:
# inspect whether app creation happens per-process in fastapi/uvicorn source code on this env,
# and whether BackgroundScheduler uses a multiprocessing-safe lock (it should not).
import ast
from pathlib import Path
for p in [Path(P) for P in ["/tmp/venv/lib/python3.12/site-packages/fastapi/applications.py", "/tmp/llc/lib/python3.12/site-packages/fastapi/applications.py"]]:
if p.exists():
print("fastapi path", p)
src = p.read_text()
print(src[src.find("class FastAPI"):src.find("class FastAPI")+1200])
try:
import APScheduler.BackgroundScheduler
import APScheduler.schedulers.base
print("apscheduler available")
# Search for start lock/execution_lock declarations in available APScheduler source
import APScheduler, inspect
src = inspect.getsource(APScheduler.schedulers.base.BaseScheduler)
print("has start_lock:", "start_lock" in src or "execution_lock" in src)
line = next((src.splitlines()[i] for i,l in enumerate(src.splitlines()) if "start_lock" in l or "execution_lock" in l), None)
print("lock token:", line)
except Exception as e:
print("apscheduler probe unavailable:", e)
print("fastapi is uvicorn mounted by lifecycle manager in known upstream implementations; process-local lifecycle starts app per worker unless external single-instance control exists.")
PYRepository: genesis-kb/transcription_engine
Length of output: 407
Run ingestion scheduling as a singleton.
The scheduler is started in every FastAPI lifespan without process/election coordination, so multiple Uvicorn workers or reload processes can schedule and run the same ingestion_pipeline job concurrently.
app/scheduler.py#L24-L43: enforce global ownership with an external scheduler or distributed lock.server.py#L15-L21: avoid starting the scheduler in every web worker unless one is elected and globally gated.
📍 Affects 2 files
app/scheduler.py#L24-L43(this comment)server.py#L15-L21
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/scheduler.py` around lines 24 - 43, Make ingestion scheduling globally
singleton-owned: update app/scheduler.py’s start_scheduler to acquire an
external scheduler or distributed lock before creating or starting the
BackgroundScheduler, and release ownership during shutdown. Update server.py’s
FastAPI lifespan startup so workers only start the scheduler when elected and
globally gated; apply the coordination changes at both listed sites.
Comment on lines
+47
to
+52
| def stop_scheduler(): | ||
| """Stop the background scheduler.""" | ||
| global _scheduler | ||
| if _scheduler is not None and _scheduler.running: | ||
| logger.info("Stopping background scheduler...") | ||
| _scheduler.shutdown(wait=False) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate scheduler.py =="
fd -a 'scheduler\.py$' . || true
echo "== app/scheduler.py outline / relevant content =="
if [ -f app/scheduler.py ]; then
wc -l app/scheduler.py
sed -n '1,140p' app/scheduler.py
fi
echo "== search scheduler references =="
rg -n "stop_scheduler|start_scheduler|_scheduler|shutdown\\(wait=False\\)|run_job|ingestion|schedule|AsyncIOScheduler" -S .Repository: genesis-kb/transcription_engine
Length of output: 50387
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== server.py lifecycle =="
sed -n '1,90p' server.py
echo "== app/services/ingestion_service.py outline/content =="
if [ -f app/services/ingestion_service.py ]; then
wc -l app/services/ingestion_service.py
sed -n '1,220p' app/services/ingestion_service.py
fi
echo "== routes/ingestion.py lifecycle =="
if [ -f routes/ingestion.py ]; then
wc -l routes/ingestion.py
sed -n '1,260p' routes/ingestion.py
fi
echo "== dependency injection/lifespan references =="
rg -n "lifespan|on_event|FastAPI|Depends|BackgroundTasks|async def lifespan|stop_scheduler|start_scheduler|IngestionService\\(" app routes tests || true
echo "== APScheduler docs/source behavior probe if package installed =="
python3 - <<'PY'
import inspect
try:
from apscheduler.schedulers.background import BackgroundScheduler
print("imports ok")
for name in ("start", "shutdown", "running"):
attr = getattr(BackgroundScheduler, name, None)
print(f"{name}: {attr}, source signature: {inspect.signature(attr) if attr else None}")
print(inspect.getsource(BackgroundScheduler.shutdown))
except Exception as e:
print(type(e).__name__, e)
PYRepository: genesis-kb/transcription_engine
Length of output: 13639
🌐 Web query:
APScheduler BackgroundScheduler shutdown wait parameter wait=False behavior running jobs documentation
💡 Result:
In APScheduler, the shutdown method on a BackgroundScheduler is used to shut down the scheduler along with its associated executors and job stores [1][2]. The wait parameter controls whether the scheduler waits for currently executing jobs to finish before completing the shutdown process [3][4]. When you call scheduler.shutdown(wait=False), the scheduler shuts down its job stores and executors immediately, but it does not wait for any currently running tasks to complete [3][5]. Notably, this action does not interrupt or terminate the currently running jobs; they continue to execute until they finish, even though the scheduler itself has initiated its shutdown sequence [1][2][6]. By default, the wait parameter is set to True, which causes the scheduler to wait until all currently executing jobs have finished before completing the shutdown [3][4].
Citations:
- 1: https://apscheduler.readthedocs.io/en/stable/modules/schedulers/background.html
- 2: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html?highlight=maximum
- 3: https://apscheduler.readthedocs.io/en/stable/userguide.html
- 4: https://apscheduler.readthedocs.io/en/3.x/userguide.html
- 5: https://apscheduler.readthedocs.io/en/latest/userguide.html
- 6: https://apscheduler.readthedocs.io/en/3.x/modules/schedulers/background.html
Wait for in-flight ingestion jobs during shutdown.
stop_scheduler() is called from FastAPI lifetime teardown, but shutdown(wait=False) leaves any running _run_ingestion_job() job to continue while IngestionService.run_full_pipeline() may still access the database or POST to the internal /transcription/add_to_queue/ endpoint. Use wait=True or coordinate cancellation and make the job cancellation-safe.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/scheduler.py` around lines 47 - 52, Update stop_scheduler() to wait for
running scheduler jobs to finish before returning by using blocking shutdown
behavior. Ensure the existing _run_ingestion_job() execution completes before
FastAPI teardown proceeds, while preserving the current scheduler-running guard
and log message.
Comment on lines
+119
to
+122
| last_scanned = None # Ignore last_scanned to fetch older videos | ||
|
|
||
| if not channel_yt_id: | ||
| logger.error(f"Channel {channel['name']} is missing yt_channel_id in config.") | ||
| return 0 | ||
|
|
||
| max_results = int(settings.config.get("channel_scan_max_results", "50")) | ||
| # Fetch up to 20 recent videos, then we will take 7 new ones | ||
| max_results = 20 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Preserve the scan cursor and drain the backlog safely.
last_scanned = None makes update_channel_scanned() ineffective. Each run rereads only the newest 20 videos and inserts 7; if more than 20 videos arrive between scans, older videos can fall out of the search window permanently. Persist/use the scan boundary or page through an explicit backlog before marking the scan complete.
Also applies to: 159-160
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/services/channel_scanner.py` around lines 119 - 122, Update the channel
scanning flow around last_scanned and update_channel_scanned() to preserve and
reuse the persisted scan cursor instead of resetting it to None. Ensure each run
pages through the backlog or otherwise continues from the saved boundary until
the pending older videos are safely considered, while retaining the existing
limit of selecting up to 7 new videos before marking the scan complete.
Comment on lines
+499
to
+512
| def save_yt_comment(self, comment_data: dict) -> Optional[dict]: | ||
| """Insert a new YouTube comment record.""" | ||
| if not self.is_available: | ||
| return None | ||
| try: | ||
| with get_session() as session: | ||
| obj = YouTubeComment(**comment_data) | ||
| session.add(obj) | ||
| session.flush() | ||
| return obj.to_dict() | ||
| except Exception as e: | ||
| logger.error(f"Failed to save YouTube comment: {e}") | ||
| return None | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n -A 30 'class YouTubeComment' app/models.pyRepository: genesis-kb/transcription_engine
Length of output: 169
🏁 Script executed:
#!/bin/bash
set -e
printf 'Tracked files matching models/service/names:\n'
git ls-files | rg 'app/(models|services)/database_service.py|models.py|youtube|YouTube' || true
printf '\nLine count app/services/database_service.py:\n'
wc -l app/services/database_service.py
printf '\nRelevant save_yt_comment area:\n'
sed -n '470,530p' app/services/database_service.py
printf '\nSearch comment models/classes in tracked files:\n'
rg -n 'YouTubeComment|YouTubeChannel|YouTubeVideo|Transcript|def save_yt_comment|to_dict' . --glob '!*.pyc' --glob '!node_modules/**' | head -200Repository: genesis-kb/transcription_engine
Length of output: 20144
🏁 Script executed:
#!/bin/bash
set -e
printf 'YouTubeComment references in tracked Python files:\n'
rg -n 'YouTubeComment|YOUTUBE_COMMENT|youtube_comment|save_yt_comment|comment_id|transcript_id' app routes test --glob '*.py' || true
printf '\napp/models.py outline/comments around YouTube models:\n'
sed -n '1,220p' app/models.py
printf '\napp/services/youtube_auth.py relevant lines:\n'
wc -l app/services/youtube_auth.py
sed -n '1,260p' app/services/youtube_auth.pyRepository: genesis-kb/transcription_engine
Length of output: 15478
Add the missing YouTubeComment model/imports or map the insert fields manually.
save_yt_comment imports and instantiates YouTubeComment(**comment_data), but YouTubeComment is not defined in app/models.py, so this will fail at import/insert with an unresolved model or unexpected constructor argument.
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 509-509: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/services/database_service.py` around lines 499 - 512, Resolve the
undefined YouTubeComment used by save_yt_comment: either add and import a model
whose fields match comment_data, or replace the model construction with an
existing database model and explicit field mapping. Ensure the inserted fields
align with the schema and that save_yt_comment can instantiate and persist the
record without unresolved symbols or unexpected arguments.
Comment on lines
+571
to
+589
| def get_transcript_by_video_id(self, video_id: str) -> Optional[dict]: | ||
| """Look up a transcript by YouTube video ID (matches media_url).""" | ||
| if not self.is_available: | ||
| return None | ||
| try: | ||
| with get_session() as session: | ||
| # media_url typically contains the full YouTube URL | ||
| obj = ( | ||
| session.query(Transcript) | ||
| .filter( | ||
| Transcript.media_url.ilike(f"%{video_id}%") | ||
| ) | ||
| .first() | ||
| ) | ||
| return obj.to_dict() if obj else None | ||
| except Exception as e: | ||
| logger.error(f"Failed to look up transcript for video {video_id}: {e}") | ||
| return None | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Unescaped _/% in ILIKE pattern causes false-positive transcript matches.
video_id can legitimately contain _ (a valid YouTube ID character), which is a single-character wildcard in SQL LIKE/ILIKE. Since it isn't escaped, Transcript.media_url.ilike(f"%{video_id}%") can match an unrelated transcript whose media_url differs only where the _ substitutes for another character, and .first() then returns that wrong record — this transcript's summary could get posted as a YouTube comment on the wrong video.
🛡️ Proposed fix: escape LIKE special characters
with get_session() as session:
# media_url typically contains the full YouTube URL
+ escaped_id = video_id.replace("%", r"\%").replace("_", r"\_")
obj = (
session.query(Transcript)
.filter(
- Transcript.media_url.ilike(f"%{video_id}%")
+ Transcript.media_url.ilike(f"%{escaped_id}%", escape="\\")
)
+ .order_by(Transcript.created_at.desc())
.first()
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def get_transcript_by_video_id(self, video_id: str) -> Optional[dict]: | |
| """Look up a transcript by YouTube video ID (matches media_url).""" | |
| if not self.is_available: | |
| return None | |
| try: | |
| with get_session() as session: | |
| # media_url typically contains the full YouTube URL | |
| obj = ( | |
| session.query(Transcript) | |
| .filter( | |
| Transcript.media_url.ilike(f"%{video_id}%") | |
| ) | |
| .first() | |
| ) | |
| return obj.to_dict() if obj else None | |
| except Exception as e: | |
| logger.error(f"Failed to look up transcript for video {video_id}: {e}") | |
| return None | |
| def get_transcript_by_video_id(self, video_id: str) -> Optional[dict]: | |
| """Look up a transcript by YouTube video ID (matches media_url).""" | |
| if not self.is_available: | |
| return None | |
| try: | |
| with get_session() as session: | |
| # media_url typically contains the full YouTube URL | |
| escaped_id = video_id.replace("%", r"\%").replace("_", r"\_") | |
| obj = ( | |
| session.query(Transcript) | |
| .filter( | |
| Transcript.media_url.ilike(f"%{escaped_id}%", escape="\\") | |
| ) | |
| .order_by(Transcript.created_at.desc()) | |
| .first() | |
| ) | |
| return obj.to_dict() if obj else None | |
| except Exception as e: | |
| logger.error(f"Failed to look up transcript for video {video_id}: {e}") | |
| return None |
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 586-586: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/services/database_service.py` around lines 571 - 589, Update
get_transcript_by_video_id so video_id is escaped before constructing the ILIKE
pattern, treating underscores and percent signs as literal characters. Pass the
matching escape character through the ILIKE expression’s escape option,
preserving the existing availability check, first-match behavior, and error
handling.
Comment on lines
+7
to
+9
| To quickly fetch and process the latest video from each seeded channel without manual review, the `ChannelScanner` (`app/services/channel_scanner.py`) was modified: | ||
| - **Limited Results**: Hardcoded `max_results = 1` to only fetch the latest video. | ||
| - **Bypass Classification**: Changed the default parsed video state to `is_technical: True` and `status: "queued"` (from `"pending"`). This allows videos to bypass the LLM classification step and go straight into the transcription queue when the ingestion pipeline runs. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Synchronize the ingestion documentation with the implementation.
The scanner fetches up to 20 results and processes up to 7 new IDs, while both documents describe fetching exactly one latest video.
docs/automated_ingestion_updates.md#L7-L9: document the actual 20/7 behavior or align the code with the guide.docs/competency_test_proof.md#L27-L33: replace the one-video claim with the actual batch behavior.
📍 Affects 2 files
docs/automated_ingestion_updates.md#L7-L9(this comment)docs/competency_test_proof.md#L27-L33
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/automated_ingestion_updates.md` around lines 7 - 9, Update the ingestion
documentation to describe the implemented batch behavior: ChannelScanner fetches
up to 20 results and processes up to 7 new video IDs, rather than exactly one
latest video. Apply this correction in docs/automated_ingestion_updates.md lines
7-9 and docs/competency_test_proof.md lines 27-33, preserving the remaining
documented behavior.
Comment on lines
+43
to
+54
| ## 3. Proof of Execution | ||
|
|
||
| Below are screenshots demonstrating the successful execution of the pipeline, including the corrected transcripts and generated summaries. | ||
|
|
||
| > [!NOTE] | ||
| > *(Paste screenshots of the terminal output showing the corrected text and summaries from `test_pipeline.py`, or screenshots of the database/API responses here).* | ||
|
|
||
| ### Corrected Transcriptions | ||
| *(Insert Screenshot Here)* | ||
|
|
||
| ### Generated Summaries | ||
| *(Insert Screenshot Here)* |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Do not label placeholder content as execution proof.
The “Proof of Execution” section still contains “Insert Screenshot Here.” Attach real evidence or mark this document as an unverified template.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/competency_test_proof.md` around lines 43 - 54, Replace the “Insert
Screenshot Here” placeholders in the “Proof of Execution” section with actual
execution screenshots and supporting evidence, or explicitly mark the document
as an unverified template instead of presenting placeholders as proof.
| source venv/bin/activate | ||
| pip install -r requirements.txt | ||
| ``` | ||
| /home/parth/Cloned_repos/transcription_engine/venv/bin/python -m uvicorn server:app --host 0.0.0.0 --port 8000 |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the queueing step and remove host-specific commands.
The runbook includes a hard-coded /home/parth/... path, then Step 5 repeats starting Uvicorn instead of calling the ingestion/queue endpoint; the second path also lacks its leading /. Replace this with a portable command such as curl -X POST http://localhost:8000/ingestion/run and document the expected 10-video flow.
Also applies to: 48-54
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/local-run-and-queue.md` at line 12, Update the local run instructions to
remove hard-coded virtual-environment paths and use the portable ingestion queue
request, including the correct /ingestion/run endpoint. Replace the repeated
Uvicorn startup step with a curl POST command, and document that the request
queues the expected 10-video flow.
…nd update automated ingestion documentation
Author
|
@Sansh2356 Please review the PR |
Contributor
There was a problem hiding this comment.
4 issues found across 8 files (changes from recent commits).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="app/services/database_service.py">
<violation number="1" location="app/services/database_service.py:568">
P2: Comment text and transcript_id are silently dropped when persisting a YouTube comment to the database. The `save_yt_comment` method passes only a subset of `comment_data` fields into `ExternalPublication`, so `comment_text` and `transcript_id` are lost on every save. Previously, `YouTubeComment(**comment_data)` would store all fields. Either add a `comment_text` column to `ExternalPublication` (or a dedicated field), or stop passing unused data to avoid confusion. For now, anyone querying saved comments will find no record of what was actually posted.</violation>
<violation number="2" location="app/services/database_service.py:573">
P2: Newly posted comments have no publication timestamp, so the comments listing sorts NULL values rather than post time. Persist a timestamp when creating `ExternalPublication` (or pass YouTube's `publishedAt` through `comment_data`).</violation>
<violation number="3" location="app/services/database_service.py:656">
P2: N+1 query in `list_yt_comments`: after fetching the list of `ExternalPublication` records, the code queries `ContentItem` individually for each result to get the video ID. Since `ExternalPublication` already has a `content_item` relationship defined, you can add eager loading (e.g., `options(joinedload(ExternalPublication.content_item))`) and read `obj.content_item.external_id` directly, reducing N+1 queries to a single query with a JOIN.</violation>
</file>
<file name="app/services/yt_commenter.py">
<violation number="1" location="app/services/yt_commenter.py:194">
P3: The `_check_existing` method looks up `existing.get("posted_at")`, but the dict returned by `get_yt_comment_by_video_id` now uses `published_at` as the key (from `ExternalPublication.to_dict()`). This means `posted_at` will always be `None` when a duplicate comment is detected, losing the timestamp in the response. Use `existing.get("published_at")` instead.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| platform="youtube", | ||
| external_pub_id=comment_data.get("comment_id"), | ||
| pub_url=comment_data.get("url") or comment_data.get("pub_url"), | ||
| status=comment_data.get("status", "posted") |
Contributor
There was a problem hiding this comment.
P2: Newly posted comments have no publication timestamp, so the comments listing sorts NULL values rather than post time. Persist a timestamp when creating ExternalPublication (or pass YouTube's publishedAt through comment_data).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At app/services/database_service.py, line 573:
<comment>Newly posted comments have no publication timestamp, so the comments listing sorts NULL values rather than post time. Persist a timestamp when creating `ExternalPublication` (or pass YouTube's `publishedAt` through `comment_data`).</comment>
<file context>
@@ -553,10 +552,33 @@ def save_yt_comment(self, comment_data: dict) -> Optional[dict]:
+ platform="youtube",
+ external_pub_id=comment_data.get("comment_id"),
+ pub_url=comment_data.get("url") or comment_data.get("pub_url"),
+ status=comment_data.get("status", "posted")
+ )
session.add(obj)
</file context>
Suggested change
| status=comment_data.get("status", "posted") | |
| status=comment_data.get("status", "posted"), | |
| published_at=datetime.now(timezone.utc), |
| d = obj.to_dict() | ||
| d["comment_id"] = obj.external_pub_id | ||
| # Need to query external_id from ContentItem | ||
| item = session.query(ContentItem).filter_by(id=obj.content_item_id).first() |
Contributor
There was a problem hiding this comment.
P2: N+1 query in list_yt_comments: after fetching the list of ExternalPublication records, the code queries ContentItem individually for each result to get the video ID. Since ExternalPublication already has a content_item relationship defined, you can add eager loading (e.g., options(joinedload(ExternalPublication.content_item))) and read obj.content_item.external_id directly, reducing N+1 queries to a single query with a JOIN.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At app/services/database_service.py, line 656:
<comment>N+1 query in `list_yt_comments`: after fetching the list of `ExternalPublication` records, the code queries `ContentItem` individually for each result to get the video ID. Since `ExternalPublication` already has a `content_item` relationship defined, you can add eager loading (e.g., `options(joinedload(ExternalPublication.content_item))`) and read `obj.content_item.external_id` directly, reducing N+1 queries to a single query with a JOIN.</comment>
<file context>
@@ -608,13 +640,23 @@ def list_yt_comments(
+ d = obj.to_dict()
+ d["comment_id"] = obj.external_pub_id
+ # Need to query external_id from ContentItem
+ item = session.query(ContentItem).filter_by(id=obj.content_item_id).first()
+ d["video_id"] = item.external_id if item else None
+ res.append(d)
</file context>
| logger.warning(f"Could not find content_item for video_id {video_id}. Cannot save comment.") | ||
| return None | ||
|
|
||
| obj = ExternalPublication( |
Contributor
There was a problem hiding this comment.
P2: Comment text and transcript_id are silently dropped when persisting a YouTube comment to the database. The save_yt_comment method passes only a subset of comment_data fields into ExternalPublication, so comment_text and transcript_id are lost on every save. Previously, YouTubeComment(**comment_data) would store all fields. Either add a comment_text column to ExternalPublication (or a dedicated field), or stop passing unused data to avoid confusion. For now, anyone querying saved comments will find no record of what was actually posted.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At app/services/database_service.py, line 568:
<comment>Comment text and transcript_id are silently dropped when persisting a YouTube comment to the database. The `save_yt_comment` method passes only a subset of `comment_data` fields into `ExternalPublication`, so `comment_text` and `transcript_id` are lost on every save. Previously, `YouTubeComment(**comment_data)` would store all fields. Either add a `comment_text` column to `ExternalPublication` (or a dedicated field), or stop passing unused data to avoid confusion. For now, anyone querying saved comments will find no record of what was actually posted.</comment>
<file context>
@@ -553,10 +552,33 @@ def save_yt_comment(self, comment_data: dict) -> Optional[dict]:
+ logger.warning(f"Could not find content_item for video_id {video_id}. Cannot save comment.")
+ return None
+
+ obj = ExternalPublication(
+ content_item_id=content_item_id,
+ platform="youtube",
</file context>
| "status": "already_commented", | ||
| "video_id": video_id, | ||
| "comment_id": existing["comment_id"], | ||
| "posted_at": existing.get("posted_at"), |
Contributor
There was a problem hiding this comment.
P3: The _check_existing method looks up existing.get("posted_at"), but the dict returned by get_yt_comment_by_video_id now uses published_at as the key (from ExternalPublication.to_dict()). This means posted_at will always be None when a duplicate comment is detected, losing the timestamp in the response. Use existing.get("published_at") instead.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At app/services/yt_commenter.py, line 194:
<comment>The `_check_existing` method looks up `existing.get("posted_at")`, but the dict returned by `get_yt_comment_by_video_id` now uses `published_at` as the key (from `ExternalPublication.to_dict()`). This means `posted_at` will always be `None` when a duplicate comment is detected, losing the timestamp in the response. Use `existing.get("published_at")` instead.</comment>
<file context>
@@ -200,6 +182,19 @@ def get_comment_status(self, video_id: str) -> dict:
+ "status": "already_commented",
+ "video_id": video_id,
+ "comment_id": existing["comment_id"],
+ "posted_at": existing.get("posted_at"),
+ }
+ return None
</file context>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
#25
Description
This PR introduces an automated background polling mechanism to ingest new YouTube videos as soon as they are uploaded. Instead of relying on manual ingestion triggers, the system will now automatically check for new videos every 5 minutes, queue them for transcription, and ultimately post the AI-generated summary as a YouTube comment.
How it works step-by-step
apscheduler) spins up in an isolated thread.IngestionService.run_full_pipeline().YouTubeCommenterServiceautomatically formats the summary and pushes the comment to YouTube using the official Google API.Changes Made
requirements.txt:apscheduler==3.10.4to handle robust background task scheduling without blocking the main asynchronous event loop.app/scheduler.py[NEW]:BackgroundSchedulerwith anIntervalTriggerset to 5 minutes, linking it directly to theIngestionService.server.py:@asynccontextmanagerlifespanevent. This ensures the background scheduler safely starts on server boot and stops gracefully during server shutdown.Summary by cubic
Automates YouTube ingestion every 5 minutes and auto-posts transcript summaries as YouTube comments on new videos. Adds a background scheduler, an OAuth-based commenter with safer error handling and DB persistence, and
/yt-botroutes.New Features
apschedulerstarts on server boot (FastAPI lifespan) and runs ingestion every 5 minutes.ExternalPublication./yt-botroutes:POST /comment,POST /comment-with-summary,GET /comments,GET /comments/{video_id},DELETE /comments/{video_id}.is_technical: Trueandstatus: "queued".summarize: True,correct: True,llm_provider: "google",diarize: True,username, and a 30s request timeout./ingestion/runis now sync to avoid event loop deadlocks. Added unit tests and docs.Migration
YOUTUBE_CLIENT_SECRETS_FILEandYOUTUBE_OAUTH_TOKEN_FILE; placeclient_secrets.jsonand runscripts/setup_youtube_oauth.pyto createyt_oauth_token.json(both git-ignored).apscheduler,google-auth-oauthlib,google-auth-httplib2.Written for commit edbcb0c. Summary will update on new commits.
Summary by CodeRabbit
New Features
Improvements
Documentation
Bug Fixes