Dự án thực chiến

import argparse
from pathlib import Path

def tao_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="don-thu-muc",
        description="Sắp xếp tệp trong thư mục theo phần mở rộng.",
        epilog="Ví dụ: don-thu-muc ~/Downloads --apply",
    )
    # Tham số bắt buộc (positional)
    p.add_argument("thu_muc", type=Path, help="Thư mục cần dọn")

    # Cờ bật/tắt
    p.add_argument("--apply", action="store_true",
                   help="Thực thi thật (mặc định chỉ xem trước)")
    p.add_argument("-r", "--de-quy", action="store_true",
                   help="Xử lý cả thư mục con")

    # Tham số có giá trị, có mặc định
    p.add_argument("--kich-thuoc-toi-thieu", type=int, default=0,
                   metavar="BYTE", help="Bỏ qua tệp nhỏ hơn (mặc định: 0)")

    # Chỉ cho phép vài giá trị
    p.add_argument("--theo", choices=["duoi", "ngay", "kich-thuoc"],
                   default="duoi", help="Tiêu chí nhóm (mặc định: duoi)")

    # Nhận nhiều giá trị
    p.add_argument("--bo-qua", nargs="*", default=[],
                   metavar="DUOI", help="Các đuôi tệp bỏ qua, vd: .tmp .log")

    # Nhóm loại trừ lẫn nhau
    nhom = p.add_mutually_exclusive_group()
    nhom.add_argument("-v", "--chi-tiet", action="store_true")
    nhom.add_argument("-q", "--im-lang", action="store_true")

    p.add_argument("--version", action="version", version="%(prog)s 1.0")
    return p

$ don-thu-muc --help
usage: don-thu-muc [-h] [--apply] [-r] [--kich-thuoc-toi-thieu BYTE]
                   [--theo {duoi,ngay,kich-thuoc}] [--bo-qua [DUOI ...]]
                   [-v | -q] [--version] thu_muc

Sắp xếp tệp trong thư mục theo phần mở rộng.

positional arguments:
  thu_muc               Thư mục cần dọn

options:
  -h, --help            show this help message and exit
  --apply               Thực thi thật (mặc định chỉ xem trước)
  ...

Ví dụ: don-thu-muc ~/Downloads --apply

import argparse, sys
from pathlib import Path

def lenh_dong_bo(args: argparse.Namespace) -> int:
    print(f"Đồng bộ {args.nguon} → {args.dich}")
    return 0

def lenh_bao_cao(args: argparse.Namespace) -> int:
    print(f"Báo cáo dạng {args.dinh_dang}")
    return 0

def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(prog="cong-cu")
    sub = p.add_subparsers(dest="lenh", required=True)

    p_db = sub.add_parser("dong-bo", help="Đồng bộ hai thư mục")
    p_db.add_argument("nguon", type=Path)
    p_db.add_argument("dich", type=Path)
    p_db.set_defaults(ham=lenh_dong_bo)      # ⭐ gắn hàm xử lý vào parser con

    p_bc = sub.add_parser("bao-cao", help="Sinh báo cáo")
    p_bc.add_argument("--dinh-dang", choices=["json", "csv", "md"], default="md")
    p_bc.set_defaults(ham=lenh_bao_cao)

    args = p.parse_args(argv)
    return args.ham(args)                    # điều phối không cần if/elif

if __name__ == "__main__":
    sys.exit(main())

"""Logic thuần — không biết gì về dòng lệnh, nên dễ test."""
from __future__ import annotations

import logging
from collections import Counter
from dataclasses import dataclass
from pathlib import Path

log = logging.getLogger(__name__)

NHOM: dict[str, frozenset[str]] = {
    "anh":      frozenset({".jpg", ".jpeg", ".png", ".gif", ".webp", ".heic"}),
    "tai_lieu": frozenset({".pdf", ".docx", ".xlsx", ".pptx", ".txt", ".md"}),
    "nen":      frozenset({".zip", ".tar", ".gz", ".rar", ".7z"}),
    "video":    frozenset({".mp4", ".mkv", ".mov", ".avi"}),
    "code":     frozenset({".py", ".js", ".ts", ".java", ".go", ".rs"}),
}

@dataclass(frozen=True)
class KeHoach:
    """Một thao tác di chuyển dự kiến."""
    nguon: Path
    dich: Path

def phan_loai(duoi: str) -> str:
    duoi = duoi.lower()
    for ten_nhom, tap_duoi in NHOM.items():
        if duoi in tap_duoi:
            return ten_nhom
    return "khac"

def ten_khong_trung(dich: Path) -> Path:
    """bao_cao.pdf → bao_cao (1).pdf → bao_cao (2).pdf ..."""
    if not dich.exists():
        return dich
    for i in range(1, 1000):
        ung_vien = dich.with_name(f"{dich.stem} ({i}){dich.suffix}")
        if not ung_vien.exists():
            return ung_vien
    raise FileExistsError(f"Quá nhiều bản trùng tên với {dich.name}")

def lap_ke_hoach(
    thu_muc: Path,
    *,
    de_quy: bool = False,
    bo_qua: frozenset[str] = frozenset(),
    kich_thuoc_toi_thieu: int = 0,
) -> list[KeHoach]:
    """Tính TOÀN BỘ việc cần làm mà không đụng vào đĩa."""
    if not thu_muc.is_dir():
        raise NotADirectoryError(f"Không phải thư mục: {thu_muc}")

    ke_hoach: list[KeHoach] = []
    da_dat: set[Path] = set()          # tên đã "đặt gạch" trong lần chạy này
    nguon_iter = thu_muc.rglob("*") if de_quy else thu_muc.iterdir()

    for f in sorted(nguon_iter):
        if not f.is_file() or f.suffix.lower() in bo_qua:
            continue
        if f.stat().st_size < kich_thuoc_toi_thieu:
            continue
        nhom = phan_loai(f.suffix)
        if f.parent.name == nhom:      # đã đúng chỗ rồi
            continue

        dich = ten_khong_trung(thu_muc / nhom / f.name)
        while dich in da_dat:            # tránh 2 tệp cùng nhắm 1 đích
            dich = ten_khong_trung(dich)
        da_dat.add(dich)
        ke_hoach.append(KeHoach(f, dich))

    return ke_hoach

def thuc_thi(ke_hoach: list[KeHoach]) -> Counter[str]:
    """Thực hiện kế hoạch. Một tệp lỗi không làm dừng cả lô."""
    thong_ke: Counter[str] = Counter()
    for kh in ke_hoach:
        try:
            kh.dich.parent.mkdir(parents=True, exist_ok=True)
            kh.nguon.rename(kh.dich)
            thong_ke[kh.dich.parent.name] += 1
            log.debug("đã chuyển %s → %s", kh.nguon.name, kh.dich)
        except OSError as e:
            thong_ke["loi"] += 1
            log.warning("bỏ qua %s: %s", kh.nguon.name, e)
    return thong_ke

"""Lớp CLI — chỉ lo phân tích tham số, in ấn, và mã thoát."""
import argparse, logging, sys
from pathlib import Path

from don_thu_muc.core import lap_ke_hoach, thuc_thi

def cau_hinh_log(chi_tiet: bool, im_lang: bool) -> None:
    muc = logging.DEBUG if chi_tiet else (logging.ERROR if im_lang else logging.INFO)
    logging.basicConfig(level=muc, format="%(levelname)s %(message)s")

def main(argv: list[str] | None = None) -> int:
    args = tao_parser().parse_args(argv)
    cau_hinh_log(args.chi_tiet, args.im_lang)

    try:
        ke_hoach = lap_ke_hoach(
            args.thu_muc,
            de_quy=args.de_quy,
            bo_qua=frozenset(d.lower() for d in args.bo_qua),
            kich_thuoc_toi_thieu=args.kich_thuoc_toi_thieu,
        )
    except (NotADirectoryError, PermissionError) as e:
        print(f"Lỗi: {e}", file=sys.stderr)
        return 2                     # mã 2 = sai cách dùng

    if not ke_hoach:
        print("Không có gì để dọn.")
        return 0

    if not args.apply:
        print(f"[XEM TRƯỚC] {len(ke_hoach)} tệp sẽ được chuyển:\n")
        for kh in ke_hoach[:20]:
            print(f"  {kh.nguon.name}  →  {kh.dich.parent.name}/{kh.dich.name}")
        if len(ke_hoach) > 20:
            print(f"  ... và {len(ke_hoach) - 20} tệp nữa")
        print("\nThêm --apply để thực hiện.")
        return 0

    thong_ke = thuc_thi(ke_hoach)
    for nhom, n in sorted(thong_ke.items()):
        print(f"  {nhom}: {n}")
    return 1 if thong_ke["loi"] else 0

if __name__ == "__main__":
    sys.exit(main())

$ uv add httpx beautifulsoup4 lxml
# httpx: HTTP client hiện đại (đồng bộ + bất đồng bộ, HTTP/2)
# beautifulsoup4: phân tích HTML
# lxml: bộ phân tích nhanh cho BeautifulSoup

from __future__ import annotations

import logging, time
from dataclasses import dataclass
from urllib.parse import urljoin, urlparse
from urllib.robotparser import RobotFileParser

import httpx
from bs4 import BeautifulSoup

log = logging.getLogger(__name__)

USER_AGENT = "HocPythonBot/1.0 (+https://vidu.vn/lien-he)"

@dataclass(frozen=True)
class BaiViet:
    tieu_de: str
    lien_ket: str
    tom_tat: str


class ThuThapLoi(Exception):
    """Lỗi gốc của module thu thập."""

class BiCam(ThuThapLoi):
    """robots.txt không cho phép truy cập URL này."""


def duoc_phep(url: str) -> bool:
    """Hỏi robots.txt trước khi lấy bất kỳ trang nào."""
    goc = urlparse(url)
    rp = RobotFileParser()
    rp.set_url(f"{goc.scheme}://{goc.netloc}/robots.txt")
    try:
        rp.read()
    except OSError:
        log.warning("Không đọc được robots.txt — mặc định KHÔNG cho phép")
        return False          # mặc định an toàn
    return rp.can_fetch(USER_AGENT, url)


def tai_trang(client: httpx.Client, url: str, *, so_lan: int = 3) -> str:
    """Tải HTML với retry + backoff luỹ thừa, tôn trọng Retry-After."""
    if not duoc_phep(url):
        raise BiCam(f"robots.txt chặn: {url}")

    loi: list[Exception] = []
    for lan in range(so_lan):
        try:
            r = client.get(url, timeout=10.0)
            if r.status_code == 429:              # quá nhiều request
                cho = float(r.headers.get("Retry-After", 5))
                log.info("Bị giới hạn, chờ %.1fs", cho)
                time.sleep(cho)
                continue
            r.raise_for_status()                  # ném với 4xx/5xx
            return r.text
        except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
            loi.append(e)
            cho = 2 ** lan                        # 1s, 2s, 4s
            log.warning("Lần %d thất bại (%s), thử lại sau %ds", lan + 1, e, cho)
            time.sleep(cho)

    raise ThuThapLoi(f"Không tải được {url} sau {so_lan} lần") from ExceptionGroup(
        "mọi lần thử đều hỏng", loi
    )


def phan_tich(html: str, url_goc: str) -> list[BaiViet]:
    """Tách dữ liệu — hàm THUẦN, test được bằng HTML mẫu offline."""
    soup = BeautifulSoup(html, "lxml")
    ket_qua: list[BaiViet] = []

    for the in soup.select("article.post"):        # CSS selector
        a = the.select_one("h2 a")
        p = the.select_one("p.summary")
        if a is None:                          # HTML có thể thiếu — đừng giả định
            continue
        ket_qua.append(BaiViet(
            tieu_de=a.get_text(strip=True),
            lien_ket=urljoin(url_goc, a["href"]),    # biến /bai/1 thành URL đầy đủ
            tom_tat=p.get_text(strip=True) if p else "",
        ))
    return ket_qua


def thu_thap(urls: list[str], *, nghi: float = 1.0) -> list[BaiViet]:
    tat_ca: list[BaiViet] = []
    with httpx.Client(headers={"User-Agent": USER_AGENT},
                     follow_redirects=True) as client:
        for i, url in enumerate(urls):
            if i:
                time.sleep(nghi)               # ⭐ lịch sự với server
            try:
                tat_ca.extend(phan_tich(tai_trang(client, url), url))
            except ThuThapLoi as e:
                log.error("Bỏ qua %s: %s", url, e)
    return tat_ca

from __future__ import annotations

import logging, os, time
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import datetime

import httpx

log = logging.getLogger(__name__)

@dataclass(frozen=True, slots=True)
class Repo:
    ten: str
    sao: int
    ngon_ngu: str | None
    cap_nhat: datetime

    @classmethod
    def tu_json(cls, d: dict) -> Repo:
        """Alternative constructor — biên giới giữa JSON tuỳ tiện và kiểu chặt chẽ."""
        return cls(
            ten=d["full_name"],
            sao=d["stargazers_count"],
            ngon_ngu=d.get("language"),
            cap_nhat=datetime.fromisoformat(d["updated_at"]),
        )


class GitHubClient:
    GOC = "https://api.github.com"

    def __init__(self, token: str | None = None) -> None:
        # ⭐ Khoá bí mật đọc từ biến môi trường, KHÔNG BAO GIỜ ghi trong code
        token = token or os.getenv("GITHUB_TOKEN")
        headers = {
            "Accept": "application/vnd.github+json",
            "User-Agent": "HocPython/1.0",
        }
        if token:
            headers["Authorization"] = f"Bearer {token}"
        self._client = httpx.Client(headers=headers, timeout=15.0)

    # --- Context manager: dùng được với `with` ---
    def __enter__(self) -> GitHubClient:
        return self

    def __exit__(self, *_) -> bool:
        self._client.close()
        return False          # không nuốt exception

    def _get(self, duong_dan: str, **tham_so) -> httpx.Response:
        """GET có retry — CHỈ retry lỗi tạm thời."""
        for lan in range(4):
            r = self._client.get(f"{self.GOC}{duong_dan}", params=tham_so)

            # Hết hạn mức → chờ tới lúc reset
            if r.status_code == 403 and r.headers.get("X-RateLimit-Remaining") == "0":
                reset = int(r.headers["X-RateLimit-Reset"])
                cho = max(reset - time.time(), 0) + 1
                log.warning("Hết hạn mức, chờ %.0fs", cho)
                time.sleep(cho)
                continue

            # 5xx = lỗi phía server → đáng retry
            if r.status_code >= 500:
                time.sleep(2 ** lan)
                continue

            # 4xx khác = lỗi phía TA → retry vô nghĩa, ném ngay
            r.raise_for_status()
            return r

        raise RuntimeError(f"GET {duong_dan} thất bại sau 4 lần")

    def tim_repo(self, tu_khoa: str, *, gioi_han: int = 100) -> Iterator[Repo]:
        """Generator — tự lo phân trang, người dùng chỉ việc lặp."""
        trang, da_lay = 1, 0
        while da_lay < gioi_han:
            r = self._get("/search/repositories",
                          q=tu_khoa, sort="stars", per_page=30, page=trang)
            muc = r.json()["items"]
            if not muc:
                return                       # hết dữ liệu
            for m in muc:
                if da_lay >= gioi_han:
                    return
                yield Repo.tu_json(m)
                da_lay += 1
            trang += 1


# --- Dùng ---
with GitHubClient() as gh:
    for repo in gh.tim_repo("language:python topic:cli", gioi_han=50):
        print(f"{repo.sao:6,}  {repo.ten}")

"""Lưu trữ bằng sqlite3 — database trong một tệp, có sẵn trong stdlib."""
from __future__ import annotations

import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager
from datetime import date
from pathlib import Path

LUOC_DO = """
CREATE TABLE IF NOT EXISTS anh_chup (
    ngay      TEXT NOT NULL,
    ten_repo  TEXT NOT NULL,
    sao       INTEGER NOT NULL,
    PRIMARY KEY (ngay, ten_repo)
);
CREATE INDEX IF NOT EXISTS idx_repo ON anh_chup(ten_repo);
"""

@contextmanager
def mo_kho(duong_dan: Path) -> Iterator[sqlite3.Connection]:
    """Context manager: tự commit khi xong, rollback khi lỗi, luôn đóng."""
    duong_dan.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(duong_dan)
    conn.row_factory = sqlite3.Row       # truy cập cột theo TÊN thay vì chỉ số
    try:
        conn.executescript(LUOC_DO)
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise                            # ném lại — đừng nuốt
    finally:
        conn.close()

def luu_anh_chup(conn: sqlite3.Connection, ngay: date,
                 du_lieu: list[tuple[str, int]]) -> int:
    cur = conn.executemany(
        "INSERT OR REPLACE INTO anh_chup (ngay, ten_repo, sao) VALUES (?, ?, ?)",
        [(ngay.isoformat(), ten, sao) for ten, sao in du_lieu],
    )                                    # ⭐ (?, ?, ?) — KHÔNG BAO GIỜ nối chuỗi SQL
    return cur.rowcount

def thay_doi(conn: sqlite3.Connection, hom_nay: date,
              hom_truoc: date) -> list[sqlite3.Row]:
    """So sánh hai ngày, trả về repo tăng sao nhiều nhất."""
    return conn.execute(
        """
        SELECT a.ten_repo, b.sao AS truoc, a.sao AS nay, a.sao - b.sao AS tang
        FROM   anh_chup a
        JOIN   anh_chup b ON a.ten_repo = b.ten_repo
        WHERE  a.ngay = ? AND b.ngay = ? AND a.sao > b.sao
        ORDER  BY tang DESC
        LIMIT  20
        """,
        (hom_nay.isoformat(), hom_truoc.isoformat()),
    ).fetchall()

import logging, sys
from datetime import date, timedelta
from logging.handlers import RotatingFileHandler
from pathlib import Path

from api_client.github import GitHubClient
from theo_doi.kho import mo_kho, luu_anh_chup, thay_doi

THU_MUC = Path(__file__).resolve().parent
KHO     = THU_MUC / "du_lieu" / "theo_doi.db"
NHAT_KY = THU_MUC / "nhat_ky" / "chay.log"

def cau_hinh_log() -> None:
    """Log ra CẢ tệp lẫn màn hình — cron không có màn hình để đọc."""
    NHAT_KY.parent.mkdir(parents=True, exist_ok=True)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
        handlers=[
            RotatingFileHandler(NHAT_KY, maxBytes=5*1024*1024, backupCount=3),
            logging.StreamHandler(sys.stderr),
        ],
    )

def main() -> int:
    cau_hinh_log()
    log = logging.getLogger("theo_doi")
    hom_nay = date.today()

    try:
        with GitHubClient() as gh:
            du_lieu = [(r.ten, r.sao)
                       for r in gh.tim_repo("language:python", gioi_han=100)]
        log.info("Lấy được %d repo", len(du_lieu))

        with mo_kho(KHO) as conn:
            luu_anh_chup(conn, hom_nay, du_lieu)
            tang = thay_doi(conn, hom_nay, hom_nay - timedelta(days=1))

        if tang:
            print(f"\n📈 Tăng sao nhiều nhất ({hom_nay}):")
            for h in tang[:10]:
                print(f"  +{h['tang']:<5} {h['ten_repo']}")
        else:
            print("Chưa có dữ liệu hôm trước để so sánh.")
        return 0

    except KeyboardInterrupt:
        log.info("Người dùng dừng")
        return 130
    except Exception:
        log.exception("Chạy thất bại")     # ⭐ log.exception giữ traceback
        return 1

if __name__ == "__main__":
    sys.exit(main())

# Linux / macOS — cron
$ crontab -e

# Chạy 8:00 sáng mỗi ngày. LƯU Ý 3 ĐIỂM:
#  1. Đường dẫn TUYỆT ĐỐI tới python trong venv — cron không biết venv
#  2. cd vào thư mục dự án trước — cron chạy ở $HOME
#  3. Chuyển hướng output — cron gửi mail nếu có stdout/stderr
0 8 * * * cd /home/anh/du-an && /home/anh/du-an/.venv/bin/python -m theo_doi >> /home/anh/du-an/cron.log 2>&1

# Kiểm tra trước khi tin: chạy thủ công với môi trường TRỐNG
$ env -i /home/anh/du-an/.venv/bin/python -m theo_doi

[project]
name = "theo-doi"
version = "1.0.0"
description = "Theo dõi repo GitHub hằng ngày"
requires-python = ">=3.12"
dependencies = ["httpx>=0.27", "python-dotenv>=1.0"]

[project.optional-dependencies]
dev = ["pytest>=8", "pytest-cov", "mypy>=1.11", "ruff>=0.6", "respx"]

[project.scripts]
theo-doi = "theo_doi.__main__:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.ruff]
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "SIM", "BLE", "C4", "PTH"]
# BLE = blind-except, PTH = ép dùng pathlib thay os.path

[tool.mypy]
python_version = "3.12"
strict = true

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--cov=src --cov-report=term-missing"