This is a submission for the Hacktoberfest Open-Source AI Challenge Week 2: Side Hustle
What I Built
Room Reader is a single-file Python tool for the person who runs a small YouTube channel as a side hustle.
You give it a channel. It reads the last few videos, pulls the comments, and hands you back a short queue of the comments that still deserve an answer: the real questions, from real viewers, that nobody ever replied to.
Not "engagement analytics." Not a dashboard. One list you can clear in ten minutes:
ROOM READER - a small channel
4 videos | 90 comments on them | scanned 41 | 1 still want a reply
REPLY QUEUE (open-weight model, offline)
1. [question 0.92 ? unanswered] @cindysharon4215 (1l):
...is pumpkin oak good and strong? Couldn't find video.
It is the tool I wish existed while I spent weeks auditing small creators' comment sections by hand. Those sections are where a side hustle actually converts: someone asks "where do I buy this," the owner hearts it, and moves on. The heart costs nothing. The missed answer costs a sale.
Room Reader answers three questions about a channel in about ten seconds:
- Which comments are actually questions? An open-weight model does this offline, no API, no labels, no training.
- Which of those did the owner already answer? It opens the reply thread under each comment and checks for the channel owner. This one mattered more than I expected (more below).
- What's left? A ranked queue, newest channels first, ready to work through.
Demo
Real run, real channel. I scanned four recent videos on a small wax-melt channel (a few hundred views each, 90 comments total). The tool read 41 of them, flagged the questions, then cleared the ones the owner had already answered in the thread.
One survivor:
@cindysharon4215: On different note is pumpkin oak good and strong? Couldn't find video.
That is a buyer, mid-sentence, asking the owner a direct question about a product. It had one like and no reply. That is exactly the comment a side hustle cannot afford to lose, and exactly the comment a tired creator scrolls past.
The same run found two other "question" candidates and dropped both, because the owner had in fact replied to them further down the thread.
Code
Single file, no build step, one dependency tree (onnxruntime, tokenizers, numpy).
Full source: https://pub-a941bfd863a24f91a60e6c4979c18a84.r2.dev/pi-sandbox-uploads/343582369728434176/2026-10-13/1791863487662-64618324-8937-41b2-83e5-266ec4b92110-roomreader.py (mirror: https://paste.rs/dmVIA)
python3 roomreader.py @yourhandle --videos 5 --comments 60 --check-replies --json out.json
The heart of it is one function:
QUESTION_HYPOTHESIS = "The viewer is asking a question."
def question_prob(self, comment):
enc = self.tok.encode(comment or " ", QUESTION_HYPOTHESIS)
n = min(len(enc.ids), MAX_LEN)
ids = np.array([(enc.ids + [0] * (MAX_LEN - n))[:MAX_LEN]], dtype=np.int64)
mask = np.array([([1] * n + [0] * (MAX_LEN - n))[:MAX_LEN]], dtype=np.int64)
logits = self.sess.run(None, {"input_ids": ids, "attention_mask": mask})[0][0]
ex = np.exp(logits - logits.max())
return float((ex / ex.sum())[ENTAILMENT])
The full file, including the YouTube/Invidious readers and the reply-thread check, is at the link above and pasted at the end of this post.
How I Built It
The model does the work, and it runs on the CPU. No hosted AI is involved. Room Reader loads Xenova/nli-deberta-v3-xsmall, a DeBERTa-v3 model fine-tuned on MNLI and exported to quantized ONNX (~87 MB). It's a cross-encoder natural language inference model: you feed it a premise and a hypothesis, and it scores whether the premise entails the hypothesis. Running that as a classifier is the classic zero-shot trick. For every comment I ask one question:
premise: <the viewer's comment>
hypothesis: "The viewer is asking a question."
and read the entailment probability. No labels, no fine-tuning, no training data of my own. The model was never trained on YouTube comments; it just happens to know what a question is.
I got the hypothesis wrong first, and that was the whole project. My first version used sentence embeddings (all-MiniLM-L6-v2, also open-weight, also ONNX) and compared each comment to a handful of intent anchors. It was useless: on short, casual comments everything sits between 0.85 and 0.93 cosine, so "Beautiful video!! Love it so much" and "How long does the cure take?" looked nearly identical.
I switched to the NLI cross-encoder and then made a second mistake: I softmaxed over four abstract labels like "This is a question." / "This is praise." / "This is spam." That looked fine in aggregate and was wrong in the cases that mattered, because the model would confidently read "I would die for those foxes" as a question. The fix was embarrassingly small: use one concrete sentence with an explicit subject, "The viewer is asking a question.", and read its entailment score directly instead of comparing it to competing labels. The separation went clean (every real question ≥ 0.89, every compliment ≤ 0.22 in my probes), and a cheap surface gate ("?" or an interrogative word) removes the last false positives without touching the real ones.
The reply-thread check changed how I see my own data. My older hand-audits counted a channel owner's top-level comments and called a room "silent" when that count was zero. Wrong. Most of those owners were answering inside the threads, where a top-level count never looks. To get it right, Room Reader takes the continuation token on each comment, fetches just that thread, and looks for authorIsChannelOwner. Across the four small channels I tested, most flagged questions had in fact been answered this way. The tool's honest output is often "nobody is waiting on you here" — and that is a true and useful answer too.
The plumbing is deliberately boring. Channel id comes from the handle page (externalId). Recent videos and view counts come from YouTube's public innertube endpoints. Comments come from a public Invidious instance. Ten-second runs, no keys, no quota.
Why Does Open Innovation Matter?
A side hustle cannot pay a per-call API bill to triage its own comment section. That's the whole reason Room Reader exists the way it does.
- The marginal cost is zero. Once the 87 MB model is on disk, reading a thousand comments costs electricity. A closed API that charges per token turns "check my comments every morning" into a line item, and the line item is what kills small tools. The side hustle can only afford a tool it can run forever for free.
- The data never leaves the laptop. The comments and the model both stay local. A viewer's comment is sometimes a private thing (an order that went wrong, a health question, a kid's name). Sending that to a third party to decide whether it's a question is a bad trade for a one-person business.
-
The model is a part you can swap, not a service you rent. If
nli-deberta-v3-xsmallisn't quite right for a niche, you can drop in a different open-weight model, or fine-tune one on your own labeled comments, without asking anyone's permission or waiting for a pricing change. The interesting failure I hit (praise read as a question) is fixable precisely because I can see the weights, probe them, and change the prompt around them. - Open weights let one specific person build for ten specific people. The thing I built is too small for anyone to sell as a product. Open-source AI is exactly the layer where tools that small get to exist.
Limitations (honest ones)
The classifier is a first pass, not a judge. It's tuned for questions; complaints ("my order came damaged") slip through the model and I catch them with a small regex, which is a hack, not a solution. Comments in other languages, heavy sarcasm, and inside-joke shorthand will misfire. Treat the queue as a shortlist to skim, not a verdict. The next version should fine-tune the xsmall model on a few hundred hand-labeled comments from one niche, and measure it like a classifier instead of trusting the threshold.
Full source
#!/usr/bin/env python3
"""
Room Reader - find the comments on a small YouTube channel that still need an answer.
Single file. Local-first. No API keys, no accounts, nothing sent to a hosted AI worker.
The only network calls read public data about a channel you own or are reviewing.
The read runs offline with an open-weight model:
Xenova/nli-deberta-v3-xsmall (DeBERTa-v3-xsmall, MNLI-fine-tuned, exported to ONNX)
A cross-encoder NLI model gives zero-shot classification for free: for each comment we
measure how strongly it entails the sentence "The viewer is asking a question." with no
training data and no labels of our own. A second, regex pass flags complaints the
question model misses (damaged, refund, never arrived).
Then it opens each thread the model flagged: if the creator already replied, the
comment drops out. What's left is the pile a channel owner never answers.
Usage:
python3 roomreader.py @handle
python3 roomreader.py @handle --videos 5 --check-replies --json out.json
python3 roomreader.py UCxxxxxxxxxxxxxxxxxxxxxx
Download the model once (~87 MB model + ~9 MB tokenizer):
mkdir -p model
curl -L -o model/model_quantized.onnx \
https://huggingface.co/Xenova/nli-deberta-v3-xsmall/resolve/main/onnx/model_quantized.onnx
curl -L -o model/tokenizer.json \
https://huggingface.co/Xenova/nli-deberta-v3-xsmall/resolve/main/tokenizer.json
"""
import argparse
import json
import os
import re
import time
import urllib.parse
import urllib.request
import numpy as np
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36")
YT_KEY = "AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8" # public web client key
INVIDIOUS = os.environ.get("INVIDIOUS", "https://inv.nadeko.net")
MODEL_DIR = os.environ.get("MODEL_DIR", os.path.join(os.path.dirname(os.path.abspath(__file__)), "model"))
MODEL_REPO = "Xenova/nli-deberta-v3-xsmall"
QUESTION_HYPOTHESIS = "The viewer is asking a question."
ENTAILMENT = 1 # id2label for these exports: 0=contradiction 1=entailment 2=neutral
MAX_LEN = 256
QUESTION_THRESHOLD = 0.5
SPAM_RE = re.compile(r"(?i)(https?://|www\.|t\.me/|follow me|sub4sub|sub 4 sub|"
r"check out my (channel|video|page)|free followers|link in bio|"
r"dm me|whatsapp|\+?\d{9,})")
CONCERN_RE = re.compile(r"(?i)(damag|broken|refund|never (arrived|received|got|came)|"
r"didn'?t (arrive|receive|get|come)|wrong (item|order|size)|"
r"scam|charged (me )?twice|customer service|no reply|unanswered)")
INTERROG_RE = re.compile(r"(?i)\b(what|where|when|why|how|who|which|whom|can|could|"
r"do|does|did|is|are|will|would|should|any|anyone|anybody|"
r"help|link|recipe|recommend|available|ship|shipment)\b")
def fetch(url, data=None, timeout=30):
req = urllib.request.Request(url, data=data)
req.add_header("User-Agent", UA)
req.add_header("Accept-Language", "en-US,en;q=0.9")
if data is not None:
req.add_header("Content-Type", "application/json")
return urllib.request.urlopen(req, timeout=timeout).read().decode("utf-8", "replace")
def walk(obj, key):
if isinstance(obj, dict):
for k, v in obj.items():
if k == key:
yield v
yield from walk(v, key)
elif isinstance(obj, list):
for item in obj:
yield from walk(item, key)
# ---------------------------------------------------------------- channel data
def resolve_channel(handle):
if re.fullmatch(r"UC[\w-]{20,}", handle):
return handle, handle
h = handle.strip().split("youtube.com/")[-1].split("?")[0].strip("/")
if not h.startswith("@"):
h = "@" + h
html = fetch(f"https://www.youtube.com/{h}")
m = re.search(r'"externalId":"(UC[\w-]+)"', html)
t = re.search(r'<meta property="og:title" content="([^"]+)"', html)
if not m:
raise SystemExit(f"could not resolve channel id for {handle}")
return m.group(1), (t.group(1) if t else h)
def recent_videos(cid, n):
body = {"context": {"client": {"clientName": "WEB", "clientVersion": "2.20260807.00.00",
"hl": "en", "gl": "US"}},
"browseId": cid, "params": "EgZ2aWRlb3PyBgQKAjoA"}
d = json.loads(fetch("https://www.youtube.com/youtubei/v1/browse?key=" + YT_KEY,
json.dumps(body).encode()))
out = []
for lv in walk(d, "lockupViewModel"):
vid = lv.get("contentId")
if not vid:
continue
title = None
for tt in walk(lv, "lockupMetadataViewModel"):
title = tt.get("title", {}).get("content")
break
out.append({"videoId": vid, "title": title})
if len(out) >= n:
break
return out
def player(vid):
body = {"context": {"client": {"clientName": "WEB", "clientVersion": "2.20260807.00.00",
"hl": "en", "gl": "US"}}, "videoId": vid}
j = json.loads(fetch("https://www.youtube.com/youtubei/v1/player?key=" + YT_KEY,
json.dumps(body).encode()))
vd = j.get("videoDetails", {}) or {}
return vd.get("title", ""), int(vd.get("viewCount") or 0)
def comments_page(vid, continuation=None, tries=3):
url = f"{INVIDIOUS}/api/v1/comments/{vid}?hl=en"
if continuation:
url += "&continuation=" + urllib.parse.quote(continuation)
for a in range(tries):
try:
return json.loads(fetch(url))
except Exception as e:
if a == tries - 1:
return {"comments": [], "error": str(e)[:80]}
time.sleep(1.5)
def fetch_comments(vid, cap=60):
first = comments_page(vid)
out = list(first.get("comments", []) or [])
total = first.get("commentCount")
cont = first.get("continuation")
while cont and len(out) < cap:
page = comments_page(vid, cont)
batch = page.get("comments", []) or []
if not batch:
break
out.extend(batch)
cont = page.get("continuation")
time.sleep(0.4)
return out[:cap], total, first.get("error")
def fetch_reply_authors(vid, continuation, cap=20):
try:
j = json.loads(fetch(f"{INVIDIOUS}/api/v1/comments/{vid}?hl=en&continuation="
+ urllib.parse.quote(continuation)))
except Exception:
return []
return [(c.get("author"), bool(c.get("authorIsChannelOwner")))
for c in (j.get("comments") or [])[:cap]]
# ---------------------------------------------------------------- local model
class Reader:
"""Zero-shot 'does this comment ask something?' via an open-weight NLI model. CPU only."""
def __init__(self, model_dir=MODEL_DIR):
import onnxruntime as ort
from tokenizers import Tokenizer
mp = os.path.join(model_dir, "model_quantized.onnx")
tp = os.path.join(model_dir, "tokenizer.json")
if not os.path.exists(mp) or not os.path.exists(tp):
base = f"https://huggingface.co/{MODEL_REPO}/resolve/main"
raise SystemExit(
"model files missing. Download once:\n"
f" mkdir -p {model_dir}\n"
f" curl -L -o {mp} {base}/onnx/model_quantized.onnx\n"
f" curl -L -o {tp} {base}/tokenizer.json")
self.tok = Tokenizer.from_file(tp)
self.tok.enable_truncation(max_length=MAX_LEN)
self.sess = ort.InferenceSession(mp, providers=["CPUExecutionProvider"])
self.names = [i.name for i in self.sess.get_inputs()]
def question_prob(self, comment):
enc = self.tok.encode(comment or " ", QUESTION_HYPOTHESIS)
n = min(len(enc.ids), MAX_LEN)
ids = np.array([(enc.ids + [0] * (MAX_LEN - n))[:MAX_LEN]], dtype=np.int64)
mask = np.array([([1] * n + [0] * (MAX_LEN - n))[:MAX_LEN]], dtype=np.int64)
feed = {"input_ids": ids, "attention_mask": mask}
if "token_type_ids" in self.names:
feed["token_type_ids"] = np.zeros_like(ids)
logits = self.sess.run(None, feed)[0][0]
ex = np.exp(logits - logits.max())
return float((ex / ex.sum())[ENTAILMENT])
def label_comment(reader, text):
if not text:
return "empty", 0.0
if SPAM_RE.search(text):
return "spam", 0.99
q = reader.question_prob(text)
cue = ("?" in text) or bool(INTERROG_RE.search(text))
if q >= QUESTION_THRESHOLD and cue:
return "question", q
if CONCERN_RE.search(text):
return "concern", q
return "other", q
# ---------------------------------------------------------------- report
def analyze(handle, nvids, cap, check_replies, reader):
cid, cname = resolve_channel(handle)
vids = recent_videos(cid, nvids)
report = {"channel": cname, "channelId": cid, "videos": [], "queue": [],
"model": MODEL_REPO, "hypothesis": QUESTION_HYPOTHESIS,
"generated": int(time.time())}
for v in vids:
vid = v["videoId"]
title, views = player(vid)
raw, total, err = fetch_comments(vid, cap)
cleaned = []
for c in raw:
if c.get("authorIsChannelOwner"):
continue
cleaned.append({
"author": c.get("author"),
"content": (c.get("content") or "").strip(),
"likes": c.get("likeCount") or 0,
"hearted": bool(c.get("creatorHeart")),
"replyCount": (c.get("replies") or {}).get("replyCount", 0),
"published": c.get("publishedText"),
"commentId": c.get("commentId"),
"_cont": (c.get("replies") or {}).get("continuation"),
})
for c in cleaned:
c["intent"], c["score"] = label_comment(reader, c["content"])
c["answered"] = None
if check_replies and c["intent"] in ("question", "concern") \
and c["replyCount"] > 0 and c["_cont"]:
c["answered"] = any(own for _, own in fetch_reply_authors(vid, c["_cont"]))
needed = [c for c in cleaned if c["intent"] in ("question", "concern")]
vid_rec = {"videoId": vid, "title": title or v.get("title"), "views": views,
"commentCount": total, "scanned": len(cleaned), "replyNeeded": len(needed)}
for c in needed:
if c["answered"] is True:
continue
report["queue"].append({
"videoId": vid, "videoTitle": title or v.get("title"),
"author": c["author"], "content": c["content"][:300],
"kind": c["intent"], "score": round(c["score"], 3), "likes": c["likes"],
"hearted": c["hearted"], "answered": c["answered"],
"published": c["published"], "commentId": c["commentId"]})
report["videos"].append(vid_rec)
time.sleep(1.0)
report["queue"].sort(key=lambda x: -x["score"])
report["summary"] = {
"videos": len(report["videos"]),
"commentsSeen": sum(v["commentCount"] or 0 for v in report["videos"]),
"scanned": sum(v["scanned"] for v in report["videos"]),
"queue": len(report["queue"]),
}
return report
def print_report(r):
s = r["summary"]
print(f"\nROOM READER - {r['channel']} ({r['channelId']})")
print(f"{s['videos']} videos | {s['commentsSeen']} comments on them | scanned {s['scanned']} | "
f"{s['queue']} still want a reply\n")
for v in r["videos"]:
print(f" {v['videoId']} views={v['views']} comments={v['commentCount']} "
f"reply-needed={v['replyNeeded']} {(v['title'] or '')[:48]}")
if not r["queue"]:
print("\n REPLY QUEUE: empty. Nobody is waiting on you here.")
print()
return
print(f"\n REPLY QUEUE (open-weight model, offline; hypothesis: \"{r['hypothesis']}\")")
for i, c in enumerate(r["queue"][:15], 1):
ans = "?" if c["answered"] is None else ("owner-answered" if c["answered"] else "unanswered")
heart = " hearted" if c["hearted"] else ""
print(f" {i:>2}. [{c['kind']} {c['score']:.2f}{heart} {ans}] {c['author']} ({c['likes']}l): "
f"{c['content'][:108].replace(chr(10), ' ')}")
print()
def main():
ap = argparse.ArgumentParser(description="Find the comments a small channel still needs to answer.")
ap.add_argument("channel", help="@handle, channel URL, or UC... id")
ap.add_argument("--videos", type=int, default=5)
ap.add_argument("--comments", type=int, default=60, help="max top-level comments scanned per video")
ap.add_argument("--check-replies", action="store_true",
help="open reply threads to confirm whether the owner already answered")
ap.add_argument("--json", metavar="PATH", help="also write the full report as JSON")
args = ap.parse_args()
reader = Reader()
report = analyze(args.channel, args.videos, args.comments, args.check_replies, reader)
print_report(report)
if args.json:
with open(args.json, "w") as f:
json.dump(report, f, indent=1)
print(f"wrote {args.json}")
if __name__ == "__main__":
main()
Top comments (0)