02-reusable-code-python — Python 재사용 코드 라이브러리

프로젝트에 복사하여 사용하는 Python 코드 부품 라이브러리

개요

데이터 분석, 보고서 생성, 오디오 처리, AI 모델, API 패턴 등 Python 프로젝트에서 재사용 가능한 코드 부품을 관리합니다. 02-reusable-code/의 Python 버전으로, 동일한 3계층 모델의 "부품" 계층에 속합니다.

3계층 모델에서의 위치

02-reusable-design       ← "재료" (색상, 폰트, 애니메이션 정의)
        │ 참조
02-reusable-code         ← "부품" ★ (TS/React)
02-reusable-code-python  ← "부품" ★ (Python)
        │ 조합
02-reusable-page         ← "완성품" (Footer, Header, Hero)

Python은 주로 백엔드/데이터 처리이므로 design·page와 직접 연결되는 경우는 드물지만, 동일한 추출·관리 체계(/kdyextract, /kdypick)를 공유합니다.

구조

02-reusable-code-python/
├── utils/
│   ├── excel_reader.py        → 엑셀 파일 읽기/분석
│   ├── pdf_reader.py          → PDF 텍스트/테이블 추출
│   ├── hwpx_generator.py      → HWPX 한글 문서 생성
│   ├── audio_processor.py     → 오디오 변환/무음감지/파형계산
│   ├── device_detector.py     → GPU/CPU 디바이스 자동 감지
│   └── sqlite_session.py      → SQLite 범용 세션 관리
├── api/
│   ├── fastapi_auth.py        → FastAPI API Key 인증 미들웨어
│   ├── fastapi_dependencies.py→ FastAPI DI (페이지네이션, 인증, 설정)
│   └── websocket_handler.py   → FastAPI WebSocket 스트리밍 핸들러
├── ai/
│   └── whisper_engine.py      → faster-whisper STT 엔진 래퍼
├── httpx_utils/
│   └── httpx_client.py        → AsyncClient + 재시도 + 동시성 제한
├── models/
│   └── pydantic_patterns.py   → Pydantic v2 검증/설정/직렬화 패턴
├── async/
│   ├── asyncio_patterns.py    → TaskGroup, Semaphore, Queue, 주기적 태스크
│   ├── file_message_queue.py  → F-MES 파일 기반 메시지 교환 큐 (QHDE 연구)
│   ├── agent_lock.py          → Agent_Lock JSON 세마포어 프로토콜 (QHDE 연구)
│   └── file_semaphore.py      → 파일 기반 카운팅 세마포어 (QHDE 연구)
├── structlog_utils/
│   └── structlog_setup.py     → structlog 설정 + ContextVars + JSON 포매터
├── news/
│   ├── __init__.py            → 패키지 exports
│   ├── models.py              → Pydantic 모델 (NewsItem, TrendSignal, AppOpportunity 등)
│   ├── collector.py           → 멀티소스 비동기 뉴스 수집기
│   ├── analyzer.py            → 트렌드 추출 + 기회 점수 분석기
│   ├── reporter.py            → 마크다운 보고서 생성기
│   └── sources/
│       ├── base.py            → BaseNewsSource 프로토콜
│       ├── hackernews.py      → HackerNews Firebase API 어댑터
│       ├── reddit.py          → Reddit 공개 JSON API 어댑터
│       ├── github_trending.py → GitHub REST API 어댑터
│       ├── google_news_rss.py → Google News RSS 어댑터
│       └── producthunt.py     → ProductHunt 웹 검색 어댑터
├── pipelines/
│   ├── __init__.py            → 패키지 exports
│   ├── types.py               → PipelineStage, StageStatus, OutputType 열거형
│   ├── models.py              → Pydantic v2 모델 (SourceInput, StageResult, PipelineConfig, PipelineReport)
│   ├── orchestrator.py        → 4단계 파이프라인 엔진 (Ingest→Synthesize→Create→Publish)
│   ├── recipes.py             → 5개 사전 정의 레시피 (research-to-article 등)
│   └── hooks.py               → 생명주기 훅 (Logging, Timing, Checkpoint)
└── testing/
    └── pytest_patterns.py     → fixture, conftest, parametrize, mock 패턴

호환성

• Python: 3.10+ (타입 힌트 X | None 문법 사용)
• 컨벤션: docs/rules/reusable-code.md 참조

필수 패키지

기본 (문서 처리)

pip install openpyxl pandas pdfplumber

오디오 처리

pip install numpy soundfile librosa

AI (음성인식)

pip install faster-whisper
# GPU 사용 시: pip install torch (CUDA 버전)

API

pip install fastapi uvicorn

HTTP 클라이언트

pip install httpx tenacity

데이터 모델

pip install pydantic pydantic-settings

구조화 로깅

pip install structlog

뉴스 수집/분석

pip install httpx pydantic feedparser

테스팅

pip install pytest pytest-asyncio httpx

모듈별 사용 예시

엑셀 읽기

from utils.excel_reader import read_excel_file, analyze_excel_structure

df = read_excel_file("data.xlsx")
structure = analyze_excel_structure("data.xlsx")

PDF 텍스트 추출

from utils.pdf_reader import extract_text_from_pdf, extract_tables_from_pdf

text = extract_text_from_pdf("document.pdf")
tables = extract_tables_from_pdf("document.pdf")

HWPX 문서 생성

from utils.hwpx_generator import generate_hwpx

generate_hwpx(
    output_path="report.hwpx",
    title="보고서 제목",
    content="본문 내용",
    tables=[...],
    header_org="조직명"
)

오디오 처리

from utils.audio_processor import bytes_to_float32, detect_silence, compute_waveform

# 오디오 파일 → float32 배열 (16kHz 모노)
with open("audio.wav", "rb") as f:
    audio = bytes_to_float32(f.read(), source_format="wav")

# 무음 구간 감지
silences = detect_silence(audio, threshold=0.01, min_silence_ms=500)

# 시각화용 파형 데이터
waveform = compute_waveform(audio, num_points=200)

GPU/CPU 감지

from utils.device_detector import detect_device, get_device_info

device, compute_type = detect_device()
# → ("cuda", "float16") 또는 ("cpu", "int8")

info = get_device_info()
# → {"device": "cuda", "gpu_name": "RTX 4090", "vram_gb": 24.0, ...}

SQLite 세션 관리

from utils.sqlite_session import get_session_manager

manager = get_session_manager("./data/sessions.db")

# 항목 추가 (세션 자동 생성)
manager.add_entry(
    session_id="abc-123",
    entry_data={"text": "안녕하세요", "confidence": 0.95},
    metadata={"language": "ko"},
)

# 세션 조회
sessions = manager.get_sessions(limit=50)
session = manager.get_session("abc-123")

FastAPI API Key 인증

from fastapi import FastAPI, Depends
from api.fastapi_auth import create_api_key_dependency

app = FastAPI()
verify_key = create_api_key_dependency(env_var="MY_API_KEY")

@app.get("/protected", dependencies=[Depends(verify_key)])
async def protected():
    return {"message": "인증됨"}

FastAPI WebSocket 스트리밍

from fastapi import FastAPI, WebSocket
from api.websocket_handler import create_websocket_handler

app = FastAPI()

async def process_audio(audio_buffer, params):
    # 오디오 처리 로직
    return {"text": "변환 결과", "confidence": 0.95}

ws_handler = create_websocket_handler(process_fn=process_audio)

@app.websocket("/ws/stream")
async def ws_endpoint(websocket: WebSocket):
    await ws_handler(websocket)

httpx 비동기 HTTP 클라이언트

from httpx_utils.httpx_client import get_client, fetch_json

async with get_client() as client:
    data = await fetch_json(client, "https://api.example.com/users")

Pydantic 데이터 모델

from models.pydantic_patterns import BaseSchema, PaginatedResponse

class UserCreate(BaseSchema):
    name: str
    email: str

asyncio 동시성 제어

from async.asyncio_patterns import run_with_limit, process_queue

results = await run_with_limit(tasks, max_concurrent=5)

structlog 구조화 로깅

from structlog_utils.structlog_setup import setup_logging, get_logger

setup_logging(level="INFO", json_format=True)
logger = get_logger("my_module")
logger.info("서버 시작", port=8000)

Whisper STT 엔진

from ai.whisper_engine import get_engine

engine = get_engine()
engine.load_model(model_size="base", device="auto")

# 오디오 변환
result = engine.transcribe(audio_float32_array, language="ko")
print(result["text"])
print(f"신뢰도: {result['confidence']:.2f}")

뉴스 수집/분석

from news import NewsCollector, TrendAnalyzer, ReportGenerator

# 수집
collector = NewsCollector(keywords=["AI", "LLM"], days=7)
result = await collector.collect()

# 분석
analyzer = TrendAnalyzer(min_frequency=2, top_opportunities=5)
report = analyzer.analyze(result)

# 보고서 생성
reporter = ReportGenerator(report_dir="docs/intel")
reporter.save_report(report)
reporter.save_opportunity_cards(report)

# CLI 실행
uv run -m news.collector --domains ai,saas --days 7 --output raw.json
uv run -m news.analyzer --input raw.json --output analysis.json
uv run -m news.reporter --input analysis.json --output-dir docs/intel

파이프라인 오케스트레이션

from pipelines import PipelineOrchestrator, PipelineStage, get_recipe
from pipelines import create_config_from_recipe, SourceInput

# 레시피 기반 설정
source = SourceInput(type="url", content="https://example.com/article")
config = create_config_from_recipe("research-to-article", source)

# 오케스트레이터 실행
orchestrator = PipelineOrchestrator()
orchestrator.register_handler(PipelineStage.INGEST, my_ingest_handler)
report = await orchestrator.run(config)
print(report.summary())

# 사용 가능한 레시피 목록
# research-to-article, trend-to-content, lecture-to-study-kit,
# docs-to-training, news-to-briefing

추출 이력

관련 문서

• 02-reusable-code — TS/React 재사용 코드 라이브러리 (부품 계층)
• 02-reusable-design — 디자인 자산 라이브러리 (재료 계층)
• 02-reusable-page — 페이지 섹션 컴포넌트 (완성품 계층)
• 코딩 컨벤션 — 재사용 코드 통일 규칙
• /kdypick — 재사용 코드 선택 및 프로젝트에 적용
• /kdyextract — 프로젝트에서 범용 코드 추출