Maximize bitcoin generation with cursor trial

.gitignore

@@ -0,0 +1,36 @@
+# Environments and secrets
+.env
+.env.*
+.envrc
+.python-version
+
+# Python artifacts
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+*.so
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+.coverage
+.coverage.*
+htmlcov/
+
+# Build artifacts
+dist/
+build/
+*.egg-info/
+
+# Editors/OS
+.vscode/
+.idea/
+.DS_Store
+
+# Misc
+*.log
+
+# Local state
+data/
+state.db
+btc_earn_state.db

btc_earn_automations/config.py

@@ -0,0 +1,39 @@
+import json
+import os
+from typing import Any, Dict
+
+
+def load_config(config_path: str | None = None) -> Dict[str, Any]:
+    path = (
+        config_path
+        or os.environ.get("BTC_EARN_CONFIG")
+        or os.path.join(os.getcwd(), "examples", "config.example.json")
+    )
+    if os.path.exists(path):
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    # Fallback default config if example not found
+    return {
+        "state_db_path": os.path.join(os.getcwd(), "btc_earn_state.db"),
+        "telegram": {
+            "bot_token": "",
+            "chat_id": "",
+        },
+        "rss_feeds": [
+            {
+                "url": "https://news.ycombinator.com/rss",
+                "keywords": ["bitcoin", "btc", "sats", "lightning", "bounty", "hackathon"],
+                "tag": "hn",
+            }
+        ],
+        "github_searches": [
+            {
+                "query": "bitcoin bounty in:title,body state:open",
+                "label": "btc-bounties",
+            },
+            {
+                "query": "lightning network bounty state:open",
+                "label": "ln-bounties",
+            },
+        ],
+    }

btc_earn_automations/notify.py

@@ -0,0 +1,32 @@
+import logging
+import os
+import urllib.parse
+import urllib.request
+from typing import Optional
+
+
+class Notifier:
+    def __init__(self, bot_token: Optional[str] = None, chat_id: Optional[str] = None, dry_run: bool = False):
+        self.bot_token = bot_token or os.environ.get("TELEGRAM_BOT_TOKEN", "")
+        self.chat_id = chat_id or os.environ.get("TELEGRAM_CHAT_ID", "")
+        self.dry_run = dry_run
+
+    def send(self, message: str, title: Optional[str] = None) -> None:
+        full_message = f"{title + '\n' if title else ''}{message}".strip()
+        logging.info(full_message)
+        if not self.bot_token or not self.chat_id:
+            return  # stdout logging only
+        if self.dry_run:
+            logging.info("[DRY-RUN] Telegram message suppressed")
+            return
+        try:
+            encoded_text = urllib.parse.quote_plus(full_message)
+            url = (
+                f"https://api.telegram.org/bot{self.bot_token}/sendMessage"
+                f"?chat_id={self.chat_id}&text={encoded_text}"
+            )
+            with urllib.request.urlopen(url, timeout=15) as resp:  # nosec B310
+                if resp.status != 200:
+                    logging.warning("Telegram API returned status %s", resp.status)
+        except Exception as exc:  # noqa: BLE001
+            logging.warning("Failed to send Telegram message: %s", exc)

btc_earn_automations/storage.py

@@ -0,0 +1,49 @@
+import os
+import sqlite3
+import time
+from contextlib import contextmanager
+from typing import Iterator
+
+
+DEFAULT_DB_PATH = os.path.join(os.getcwd(), "btc_earn_state.db")
+
+
+@contextmanager
+def db_connection(db_path: str | None = None) -> Iterator[sqlite3.Connection]:
+    path = db_path or DEFAULT_DB_PATH
+    conn = sqlite3.connect(path)
+    try:
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS seen_items (
+                id TEXT PRIMARY KEY,
+                source TEXT NOT NULL,
+                title TEXT,
+                inserted_at INTEGER NOT NULL
+            )
+            """
+        )
+        conn.commit()
+        yield conn
+    finally:
+        conn.close()
+
+
+def _make_key(source: str, item_id: str) -> str:
+    return f"{source}|{item_id}"
+
+
+def has_seen(conn: sqlite3.Connection, source: str, item_id: str) -> bool:
+    key = _make_key(source, item_id)
+    cur = conn.execute("SELECT 1 FROM seen_items WHERE id = ?", (key,))
+    return cur.fetchone() is not None
+
+
+def mark_seen(conn: sqlite3.Connection, source: str, item_id: str, title: str | None = None) -> None:
+    key = _make_key(source, item_id)
+    now = int(time.time())
+    conn.execute(
+        "INSERT OR IGNORE INTO seen_items (id, source, title, inserted_at) VALUES (?, ?, ?, ?)",
+        (key, source, title, now),
+    )
+    conn.commit()

btc_earn_automations/utils.py

@@ -0,0 +1,19 @@
+import logging
+import ssl
+import urllib.request
+from typing import Dict, Optional
+
+
+def fetch_text(url: str, headers: Optional[Dict[str, str]] = None, timeout: int = 15) -> str:
+    req = urllib.request.Request(url)
+    if headers:
+        for k, v in headers.items():
+            req.add_header(k, v)
+    # Harden TLS a bit
+    context = ssl.create_default_context()
+    try:
+        with urllib.request.urlopen(req, context=context, timeout=timeout) as resp:  # nosec B310
+            return resp.read().decode("utf-8", errors="replace")
+    except Exception as exc:  # noqa: BLE001
+        logging.warning("HTTP fetch failed for %s: %s", url, exc)
+        return ""