Build your own research collector (free Python lab)

$ mkdir -p ~/collector && cd ~/collector
$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install -q requests feedparser pyyaml trafilatura
$ pip list 2>/dev/null | grep -iE '^(requests|feedparser|pyyaml|trafilatura) '

feedparser         6.0.14
PyYAML             6.0.3
requests           2.34.2
trafilatura        2.3.1

$ mkdir -p notes
$ cat > notes/week1.md <<'EOF'
# Week 1: passwords and backups
Turned on a password manager and passkeys for email.
Set up a weekly backup of my documents to an external drive.
EOF
$ cat > config.yaml <<'EOF'
name: My research report
output_folder: output
thin_threshold: 0          # topics mentioned by 0 items go on the idea list

topics:                    # topic name: [keywords to look for]
  Ransomware: [ransomware]
  Microsoft: [microsoft, windows]
  Network devices: [firewall, vpn, router, cisco, fortinet, ivanti]
  Passwords: [password, passwords, passkey, passkeys, credential, credentials]
  AI: [ai, llm, chatgpt]
  Backups: [backup, backups]
  Wireless: [wi-fi, wireless, wpa3]

sources:
  - name: CISA KEV
    type: cisa_kev
    url: https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json
    max_items: 50
  - name: Schneier on Security
    type: rss
    url: https://www.schneier.com/feed/atom/
    max_items: 5
    full_text: true
  - name: My notes
    type: local_file
    path: notes
EOF
$ python -c "import yaml; c = yaml.safe_load(open('config.yaml')); print(len(c['topics']), 'topics,', len(c['sources']), 'sources')"

7 topics, 3 sources

$ cat > rc.py <<'EOF'
"""Research collector: collect, clean, store, match, report."""
import csv, hashlib, json, re, sys, time
from collections import Counter
from datetime import date
from pathlib import Path
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser

import feedparser, requests, trafilatura, yaml

UA = "ResearchCollector/1.0"           # who we say we are
session = requests.Session()
session.headers["User-Agent"] = UA
robots, last_hit = {}, {}              # remembered per website


def polite_get(url, delay=3):
    """Fetch a URL only if robots.txt allows it, and never faster than `delay` seconds per site."""
    host = "{0.scheme}://{0.netloc}".format(urlparse(url))
    if host not in robots:
        rules = RobotFileParser()
        reply = session.get(host + "/robots.txt", timeout=30)
        if reply.status_code >= 500:
            rules.disallow_all = True          # server trouble: treat as "no" for now
        elif reply.status_code >= 400:
            rules.allow_all = True             # no robots.txt to read: no rules (RFC 9309)
        else:
            rules.parse(reply.text.splitlines())
        robots[host] = rules
    if not robots[host].can_fetch(UA, url):
        print("  skipped, robots.txt says no:", url)
        return None
    wait = max(delay, robots[host].crawl_delay(UA) or 0) - (time.time() - last_hit.get(host, 0))
    if wait > 0:
        time.sleep(wait)
    last_hit[host] = time.time()
    reply = session.get(url, timeout=60)
    reply.raise_for_status()
    return reply
EOF
$ python -c "import rc; r = rc.polite_get('https://www.schneier.com/feed/atom/'); print(r.status_code, len(r.content), 'bytes')"
$ python -c "import rc; print(rc.polite_get('https://www.schneier.com/wp-admin/'))"

200 32048 bytes
  skipped, robots.txt says no: https://www.schneier.com/wp-admin/
None

$ cat >> rc.py <<'EOF'


def item(source, title, url, when, text):
    """Every source turns into the same simple shape."""
    return {"source": source, "title": title.strip(), "url": url,
            "date": str(when or ""), "text": text.strip()}


def clean_page(url):
    """Download an article and keep only its main text (no menus, ads or footers)."""
    reply = polite_get(url)
    return trafilatura.extract(reply.content, url=url) if reply is not None else None


def from_rss(src):
    reply = polite_get(src["url"])
    if reply is None:
        return []
    found = []
    for entry in feedparser.parse(reply.content).entries[: src.get("max_items", 10)]:
        t = entry.get("published_parsed")
        when = date(t.tm_year, t.tm_mon, t.tm_mday) if t else ""
        text = re.sub(r"<[^>]+>", " ", entry.get("summary", ""))   # strip HTML tags
        if src.get("full_text"):
            text = clean_page(entry.link) or text
        found.append(item(src["name"], entry.get("title", ""), entry.link, when, text))
    return found


def from_cisa_kev(src):
    reply = polite_get(src["url"])
    if reply is None:
        return []
    vulns = sorted(reply.json()["vulnerabilities"], key=lambda v: v["dateAdded"], reverse=True)
    return [item(src["name"], f'{v["cveID"]}: {v["vulnerabilityName"]}',
                 "https://nvd.nist.gov/vuln/detail/" + v["cveID"], v["dateAdded"],
                 f'{v["vendorProject"]} {v["product"]}. {v["shortDescription"]}')
            for v in vulns[: src.get("max_items", 50)]]


def from_local_file(src):
    found = []
    for f in sorted(Path(src["path"]).glob("*")):
        if f.suffix in (".txt", ".md"):
            text = f.read_text(encoding="utf-8")
            found.append(item(src["name"], text.splitlines()[0].lstrip("# "), f.resolve().as_uri(),
                              date.fromtimestamp(f.stat().st_mtime), text))
    return found


COLLECT = {"rss": from_rss, "cisa_kev": from_cisa_kev, "local_file": from_local_file}
EOF
$ python - <<'EOF'
import rc, yaml
for src in yaml.safe_load(open("config.yaml"))["sources"]:
    got = rc.COLLECT[src["type"]](src)
    first = got[0] if got else {"title": "-", "text": ""}
    print(f'{src["name"]}: {len(got)} items. First: {first["title"][:50]} ({len(first["text"])} chars)')
EOF

CISA KEV: 50 items. First: CVE-2024-27834: Apple Multiple Products Improper A (234 chars)
Schneier on Security: 5 items. First: AIs Hacking the Law (1537 chars)
My notes: 1 items. First: Week 1: passwords and backups (145 chars)

$ cat >> rc.py <<'EOF'


def slug(text):
    return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")[:60] or "item"


def save(it, library):
    """One Markdown file per item, with where it came from and when, at the top."""
    folder = Path(library) / slug(it["source"])
    folder.mkdir(parents=True, exist_ok=True)
    path = folder / (slug(it["title"]) + ".md")
    path.write_text(
        f'---\ntitle: {json.dumps(it["title"])}\nsource: {it["source"]}\nurl: {it["url"]}\n'
        f'date: {it["date"]}\ncollected: {date.today()}\n---\n\n# {it["title"]}\n\n{it["text"]}\n',
        encoding="utf-8")
    return path
EOF
$ python -c "
import rc
it = rc.from_local_file({'name': 'My notes', 'path': 'notes'})[0]
print(rc.save(it, 'output/library'))
"
$ head -8 output/library/my-notes/week-1-passwords-and-backups.md

output/library/my-notes/week-1-passwords-and-backups.md
---
title: "Week 1: passwords and backups"
source: My notes
url: file:///home/steven/collector/notes/week1.md
date: 2026-11-06
collected: 2026-11-06
---

$ cat >> rc.py <<'EOF'


def topic_patterns(topics):
    """One pattern per topic. Whole words only, any case: 'ai' will not match 'said'."""
    patterns = {}
    for name, words in topics.items():
        alts = "|".join(re.escape(w) for w in (words or [name]))
        patterns[name] = re.compile(rf"(?<![a-z0-9])(?:{alts})(?![a-z0-9])", re.IGNORECASE)
    return patterns


def tag(it, patterns):
    text = it["title"] + " " + it["text"]
    it["topics"] = [name for name, p in patterns.items() if p.search(text)]
EOF
$ python -c "
import rc
p = rc.topic_patterns({'AI': ['ai'], 'Backups': ['backup']})
for t in ['He said hello', 'New AI tool', 'Weekly backup done']:
    it = {'title': t, 'text': ''}; rc.tag(it, p); print(repr(t), '->', it['topics'])
"

'He said hello' -> []
'New AI tool' -> ['AI']
'Weekly backup done' -> ['Backups']

$ cat >> rc.py <<'EOF'


def write_report(cfg, items, out):
    new = sorted([i for i in items if i["status"] != "seen"], key=lambda i: i["date"], reverse=True)
    counts = Counter(t for i in items for t in i["topics"])
    lines = [f'# {cfg["name"]}', "", f"{len(items)} items, {len(new)} new or changed.", "",
             "## What's new", ""]
    lines += [f'- {i["date"]} [{i["title"]}]({i["url"]}) ({i["source"]})' for i in new[:15]] \
        or ["- nothing new since the last run"]
    lines += ["", "## Mentioned often", ""]
    lines += [f"- {n}: {counts[n]}" for n in sorted(cfg["topics"], key=lambda n: -counts[n])]
    lines += ["", "## Ideas: little or no material", ""]
    lines += [f"- {n}" for n in cfg["topics"] if counts[n] <= cfg.get("thin_threshold", 0)] \
        or ["- every topic has material"]
    (out / "ideas.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    with open(out / "items.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["date", "source", "title", "url", "status", "topics", "file"])
        for i in items:
            w.writerow([i["date"], i["source"], i["title"], i["url"], i["status"],
                        "; ".join(i["topics"]), i["file"]])


def run(config_file):
    cfg = yaml.safe_load(Path(config_file).read_text(encoding="utf-8"))
    out = Path(cfg["output_folder"])
    out.mkdir(parents=True, exist_ok=True)
    state_file = out / "state.json"                  # what we saw last time
    seen = json.loads(state_file.read_text()) if state_file.exists() else {}
    patterns = topic_patterns(cfg["topics"])
    items = []
    for src in cfg["sources"]:
        got = COLLECT[src["type"]](src)
        print(f'{src["name"]}: {len(got)} items')
        items += got
    for it in items:
        tag(it, patterns)
        key = it["source"] + "|" + it["url"]
        fingerprint = hashlib.sha1((it["title"] + it["text"]).encode()).hexdigest()
        it["status"] = "new" if key not in seen else "changed" if seen[key] != fingerprint else "seen"
        seen[key] = fingerprint
        it["file"] = str(save(it, out / "library"))
    state_file.write_text(json.dumps(seen, indent=0))
    write_report(cfg, items, out)
    print("Report:", out / "ideas.md")


if __name__ == "__main__":
    run(sys.argv[1] if len(sys.argv) > 1 else "config.yaml")
EOF
$ python rc.py config.yaml
$ sed -n '1,12p;/## Mentioned/,$p' output/ideas.md

# My research report

56 items, 56 new or changed.

## What's new

- 2026-11-06 [Week 1: passwords and backups](file:///home/steven/collector/notes/week1.md) (My notes)
- 2026-11-04 [CVE-2024-27834: Apple Multiple Products Improper Authentication Vulnerability](https://nvd.nist.gov/vuln/detail/CVE-2024-27834) (CISA KEV)
...
## Mentioned often

- Network devices: 9
- AI: 4
- Microsoft: 2
- Passwords: 2
- Backups: 1
- Ransomware: 0
- Wireless: 0

## Ideas: little or no material

- Ransomware
- Wireless

$ echo "Tried WPA3 on the home Wi-Fi today." >> notes/week1.md
$ python rc.py config.yaml
$ sed -n '1,8p' output/ideas.md
$ grep -c ',seen,' output/items.csv

CISA KEV: 50 items
Schneier on Security: 5 items
My notes: 1 items
Report: output/ideas.md
# My research report

56 items, 1 new or changed.

## What's new

- 2026-11-06 [Week 1: passwords and backups](file:///home/steven/collector/notes/week1.md) (My notes)

55

$ mkdir -p jobs
$ cat > jobs/post1.txt <<'EOF'
SOC Analyst (made-up sample post)
Triage alerts in a SIEM such as Splunk. Read Windows and Linux logs. PowerShell is a plus.
EOF
$ cat > jobs/post2.txt <<'EOF'
IT Support Technician (made-up sample post)
Support Windows 11 users and reset passwords. Basic SIEM and PowerShell skills are a plus.
EOF
$ cat > jobs.yaml <<'EOF'
name: Job-skills match
output_folder: output-jobs
thin_threshold: 0
topics:                    # skills you have, or are thinking about learning
  Windows: [windows]
  Linux: [linux]
  PowerShell: [powershell]
  SIEM: [siem, splunk, sentinel]
  Python: [python]
sources:
  - name: Job posts I saved
    type: local_file
    path: jobs
EOF
$ python rc.py jobs.yaml
$ sed -n '/Mentioned/,$p' output-jobs/ideas.md

Job posts I saved: 2 items
Report: output-jobs/ideas.md
## Mentioned often

- Windows: 2
- PowerShell: 2
- SIEM: 2
- Linux: 1
- Python: 0

## Ideas: little or no material

- Python

$ deactivate
$ cd ~ && rm -rf ~/collector
$ ls ~/collector

ls: cannot access '/home/steven/collector': No such file or directory