--- name: naver-news-briefing description: Search, brief, and monitor 대한민국 Naver News via the Naver Search API using natural-language Korean queries. Use when the user wants 네이버 뉴스 브리핑, 최근 N일 뉴스 요약, 제외어 포함 뉴스 검색, 여러 질의를 한 번에 묶은 멀티 브리핑, 키워드 그룹 저장/관리, 특정 키워드의 지속 감시 규칙 추가/목록/삭제, 채팅형 자동화 요청을 watch/group 설정으로 바꾸기, or cron-friendly operator guidance for periodic news checks. Prefer this skill for Korean Naver-news workflows backed by local persistent state. --- # naver-news-briefing Use the CLI script at `scripts/naver_news_briefing.py`. ## Onboarding - Treat Naver Search API credentials as mandatory before the first real use. - Tell the user early that search/briefing/watch flows will fail until `client_id` and `client_secret` are stored. - If the user has not completed setup yet, direct them to run: - `python scripts/naver_news_briefing.py setup --client-id ... --client-secret ...` - `python scripts/naver_news_briefing.py check-credentials --json` - Present setup as the first-run path, not an optional advanced step. - When helping with installation or first use, mention that credentials are stored in `data/config.json` and use DPAPI-backed secret storage on Windows when possible. ## Workflow 1. Store credentials once before any search/brief/watch command. - `python scripts/naver_news_briefing.py setup --client-id ... --client-secret ...` - Verify with `python scripts/naver_news_briefing.py check-credentials --json` 2. Run a one-shot briefing. - `python scripts/naver_news_briefing.py search "최근 3일 반도체 뉴스 브리핑 -광고"` - Add `--json` for machine-readable output. 3. Manage persistent watch rules with optional operator metadata. - Add: `python scripts/naver_news_briefing.py watch-add semiconductor "최근 7일 반도체 -광고" --label "반도체 감시" --tag watch --template watch-alert` - List: `python scripts/naver_news_briefing.py watch-list` - Remove: `python scripts/naver_news_briefing.py watch-remove semiconductor` - Check: `python scripts/naver_news_briefing.py watch-check semiconductor --json` 4. Manage persistent keyword groups for recurring briefings. - Add: `python scripts/naver_news_briefing.py group-add market-watch "최근 3일 반도체 -광고" "오늘 AI 데이터센터 -주가" --label "아침 시장" --tag 테크 --context "오전 보고용" --template morning-briefing` - List: `python scripts/naver_news_briefing.py group-list` - Inspect one group: `python scripts/naver_news_briefing.py group-list market-watch --json` - Update: `python scripts/naver_news_briefing.py group-update market-watch --add-query "배터리 공급망 -광고" --tag 공급망 --template analyst` - Remove: `python scripts/naver_news_briefing.py group-remove market-watch` 5. Run combined briefings. - `python scripts/naver_news_briefing.py brief-multi --group market-watch` - `python scripts/naver_news_briefing.py brief-multi --group market-watch --query "환율 뉴스" --template morning-briefing --json` - If `--template` is omitted, prefer the saved group template when present. 6. Convert chat-style automation requests into structured plans. - Inspect plan: `python scripts/naver_news_briefing.py plan "반도체 뉴스 1시간마다 모니터링해줘" --json` - Build an OpenClaw-friendly integration bundle: `python scripts/naver_news_briefing.py integration-plan "반도체 뉴스 1시간마다 모니터링해줘" --json` - Support practical Korean patterns such as `매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘` and `증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘`. 7. Materialize plans into persistent configs. - Watch: `python scripts/naver_news_briefing.py plan-save "반도체 뉴스 1시간마다 모니터링해줘" --as watch --name semiconductor-hourly` - Group: `python scripts/naver_news_briefing.py plan-save "매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘" --as group --name morning-tech --label "아침 브리핑"` ## Behavior - Parse positive keywords and `-제외어` using the upstream tab-search policy. - Interpret recent-news phrases such as `오늘`, `최근 3일`, `최근 2주`, `한달`, `이번주`, `지난주` as a date window and remove that phrase from the API search query. - Normalize more natural Korean sentence inputs by stripping request phrases, common 조사, duplicate tokens, and Korean exclusion phrases such as `A 말고`, `B 빼고`, `C 제외` into `-제외어` tokens. - Parse practical schedule intent from Korean chat requests into `interval`, `daily`, `weekly`, or `manual` plans. - Preserve operator-facing metadata on saved watch/group entries: `label`, `tags`, `template`, `schedule`, `operator_hints`, and original request context. - Use DPAPI-backed secret storage on Windows when possible. - Deduplicate watch notifications by `(watch_id, link)` so repeated cron runs emit only newly seen items. - `plan` returns cron-friendly operator hints, recommended commands, and a storage-target recommendation (`watch` vs `group`). - `integration-plan` returns a more practical operator bundle: save command, run command, schedule object, cron line, OpenClaw-friendly systemEvent text, and a Korean confirmation summary. - `plan-save` materializes a parsed plan into a saved `watch` or `group` configuration without owning cron wiring itself. - `brief-multi` returns chat-friendly combined text by default and structured JSON with `--json`. ## Notes - Read `references/upstream-notes.md` before major edits. - The skill uses headline/summary metadata from the Naver Search API. It does not fetch or summarize full article bodies. - Keep additions additive: preserve existing `search`, `watch-add`, `watch-list`, `watch-remove`, and `watch-check` flows. - Public user-facing documentation lives in `README.md`; keep it Korean-first and concrete. FILE:README.md # naver-news-briefing 네이버 Search API 기반으로 **뉴스 검색 / 브리핑 / 지속 감시 / 키워드 그룹 / 자동화 계획 생성**을 수행하는 OpenClaw 스킬입니다. > [!IMPORTANT] > 이 스킬은 **네이버 개발자센터에서 발급받은 Search API 자격증명(client_id, client_secret)이 있어야만 정상 동작**합니다. > 설치만으로 바로 검색되지는 않으며, **최초 1회 자격증명 입력(onboarding/setup)** 을 반드시 거쳐야 합니다. 이 스킬의 핵심은 단순 검색이 아니라 **채팅형 한국어 요청을 실제 운영 가능한 로컬 설정으로 연결**하는 데 있습니다. 즉, 사람이 이렇게 말하면: - `반도체 뉴스 1시간마다 모니터링해줘` - `매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘` - `증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘` 이 스킬은 이를 **정규화된 질의 + 저장 가능한 watch/group 설정 + cron 연결 힌트 + 추천 실행 명령**으로 바꿉니다. ## 설치 / 링크 - GitHub: - ClawHub 홈: - 설치 명령: `clawhub install naver-news-briefing` ## 대표 예시 요청 모음 - `최근 3일 반도체 뉴스 브리핑해줘` - `반도체 뉴스 1시간마다 모니터링해줘` - `매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘` - `증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘` - `환율이랑 2차전지 뉴스 묶어서 아침 브리핑용 그룹 만들어줘` --- ## 핵심 기능 - 한국어 자연어 뉴스 검색 - `-제외어` 필터링 - 최근 기간 해석 - `오늘`, `최근 3일`, `최근 2주`, `한달`, `이번주`, `지난주` - 문장형 한국어 요청 정규화 - 원샷 브리핑 - watch rule 저장 / 목록 / 삭제 / 신규 기사 체크 - 키워드 그룹 저장 / 수정 / 삭제 - 멀티 브리핑 템플릿 - `concise` - `analyst` - `morning-briefing` - `watch-alert` - 자연어 자동화 계획 파싱 - `interval / daily / weekly / manual` - 일정 설명 + cron 힌트 + 추천 템플릿 + 추천 저장 대상 - `plan-save`로 watch/group 저장 - **저장 시 운영 메타데이터까지 같이 보존** - label - tags - template - schedule - operator_hints - 기본 텍스트 출력 + `--json` 구조화 출력 --- ## 빠른 시작 ### 0) 먼저 준비할 것 이 스킬을 처음 쓰는 사용자는 먼저 아래를 준비해야 합니다. - 네이버 개발자센터에서 발급받은 `client_id` - 네이버 개발자센터에서 발급받은 `client_secret` 자격증명이 없으면 `search`, `watch-add`, `watch-check`, `brief-multi`, `plan-save` 같은 실제 운영 명령은 정상적으로 끝까지 진행되지 않습니다. ### 1) 최초 온보딩: 자격증명 저장 명령줄 인자를 한 번에 넣는 방식: ```bash python scripts/naver_news_briefing.py setup --client-id YOUR_ID --client-secret YOUR_SECRET python scripts/naver_news_briefing.py check-credentials --json ``` 더 사용자 친화적인 대화형 방식: ```bash python scripts/naver_news_briefing.py setup ``` 위처럼 `setup`만 실행하면 CLI가 `client_id`, `client_secret`를 순서대로 물어봅니다. 특히 `client_secret`는 화면에 그대로 보이지 않도록 입력됩니다. 저장 직후 실제 API까지 바로 확인하고 싶다면: ```bash python scripts/naver_news_briefing.py setup --live-check python scripts/naver_news_briefing.py setup --test-search "최근 1일 반도체 뉴스" ``` 설명: - `setup`은 최초 1회 실행하는 온보딩 명령입니다. - 입력값에 공백/줄바꿈이 섞였거나 지나치게 짧으면 저장 전에 먼저 알려줍니다. - 자격증명은 `data/config.json`에 저장됩니다. - Windows에서는 가능하면 DPAPI 기반으로 `client_secret`를 보호합니다. - `check-credentials --json`으로 첫 입력이 제대로 끝났는지 바로 검증하세요. - 일반 텍스트 확인용으로는 `python scripts/naver_news_briefing.py check-credentials` 도 사용할 수 있습니다. - 실제 API까지 확인하려면 `python scripts/naver_news_briefing.py check-credentials --live-check` 를 사용할 수 있습니다. ### 2) 원샷 브리핑 ```bash python scripts/naver_news_briefing.py search "최근 3일 반도체 뉴스 브리핑 -광고" ``` ### 3) 자연어 자동화 계획 확인 ```bash python scripts/naver_news_briefing.py plan "반도체 뉴스 1시간마다 모니터링해줘" python scripts/naver_news_briefing.py plan "매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘" --json python scripts/naver_news_briefing.py plan "증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘" ``` ### 4) OpenClaw/cron 연동 번들 생성 ```bash python scripts/naver_news_briefing.py integration-plan "반도체 뉴스 1시간마다 모니터링해줘" python scripts/naver_news_briefing.py integration-plan "매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘" --json python scripts/naver_news_briefing.py integration-plan "반도체 뉴스 1시간마다 모니터링해줘" --output data/semiconductor-hourly-bundle.json --json ``` 이 명령은 한 번에 아래를 만듭니다. - 해석된 자동화 계획 - 실제 저장 명령 (`plan-save`) - 반복 실행 명령 (`watch-check` 또는 `brief-multi`) - cron 한 줄 예시 - OpenClaw cron/systemEvent에 붙이기 좋은 텍스트 - 사용자 확인용 짧은 한국어 문구 ### 5) 계획을 실제 설정으로 저장 watch로 저장: ```bash python scripts/naver_news_briefing.py plan-save "반도체 뉴스 1시간마다 모니터링해줘" --as watch --name semiconductor-hourly ``` group으로 저장: ```bash python scripts/naver_news_briefing.py plan-save \ "매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘" \ --as group \ --name morning-tech \ --label "아침 브리핑" \ --tag 테크 \ --tag 시장 ``` --- ## 한국어 요청이 어떻게 해석되는가 ### 예시 1) `반도체 뉴스 1시간마다 모니터링해줘` 해석 포인트: - 작업 유형: `monitor` - 저장 대상 추천: `watch` - 주제 질의: `반도체` - 일정: `1시간마다` - 템플릿: `watch-alert` - 후속 실행 추천: `watch-check --json` 권장 운영: 1. `plan`으로 해석 확인 2. `plan-save --as watch` 3. 외부 스케줄러에서 `watch-check`를 1시간마다 실행 4. 신규 기사만 메신저로 전달 ### 예시 2) `매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘` 해석 포인트: - 작업 유형: `briefing` - 저장 대상 추천: `group` - 주제 질의: `반도체`, `AI 데이터센터` - 일정: `매일 07:00` - 템플릿: `morning-briefing` - 후속 실행 추천: `brief-multi --group --template morning-briefing` 권장 운영: 1. `plan`으로 schedule / grouping 확인 2. `plan-save --as group` 3. 외부 스케줄러에서 `brief-multi`를 07:00에 실행 4. stdout 텍스트를 텔레그램/디스코드/노션 등으로 전달 ### 예시 3) `증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘` 해석 포인트: - 작업 유형: `monitor` - 주제 질의: `삼성전자 -증권사 -리포트` - 일정: `15분마다`로 보수적으로 해석 - 템플릿: `watch-alert` - 제외어 유지: `증권사`, `리포트` 이 패턴은 실무에서 꽤 중요합니다. “말고 / 빼고 / 제외”가 자동으로 `-제외어`로 정규화되므로, 채팅형 요청을 곧바로 watch로 저장하기 쉽습니다. --- ## CLI 개요 ### search ```bash python scripts/naver_news_briefing.py search "최근 3일 반도체 뉴스 브리핑 -광고" python scripts/naver_news_briefing.py search "AI 데이터센터 뉴스" --json ``` ### watch ```bash python scripts/naver_news_briefing.py watch-add semiconductor "최근 7일 반도체 -광고" python scripts/naver_news_briefing.py watch-add samsung-watch "삼성전자 -증권사 -리포트" --label "삼성전자 감시" --tag watch --tag 삼성 --template watch-alert python scripts/naver_news_briefing.py watch-list python scripts/naver_news_briefing.py watch-check semiconductor --json python scripts/naver_news_briefing.py watch-remove semiconductor ``` ### group ```bash python scripts/naver_news_briefing.py group-add market-watch "최근 3일 반도체 -광고" "오늘 AI 데이터센터 -주가" --label "시장 체크" --tag 테크 --template morning-briefing python scripts/naver_news_briefing.py group-list python scripts/naver_news_briefing.py group-update market-watch --add-query "배터리 공급망 -광고" --tag 테크 --tag 공급망 --template analyst python scripts/naver_news_briefing.py group-remove market-watch ``` ### brief-multi ```bash python scripts/naver_news_briefing.py brief-multi --group market-watch --template concise python scripts/naver_news_briefing.py brief-multi --group market-watch --query "환율 뉴스" --template morning-briefing --json ``` 참고: - `--template`를 주지 않으면 group에 저장된 template를 우선 사용합니다. - 운영자가 template를 group 단위로 고정해 두면 cron 명령이 짧아집니다. ### plan / integration-plan / plan-save ```bash python scripts/naver_news_briefing.py plan "반도체 뉴스 1시간마다 모니터링해줘" python scripts/naver_news_briefing.py integration-plan "매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘" --json python scripts/naver_news_briefing.py plan-save "증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘" --as watch --name samsung-watch ``` `integration-plan`은 `plan`보다 한 단계 더 나가서 **실행 가능한 연동 번들**을 만듭니다. 대표 필드: - `storage.save_command`: 상태 저장용 `plan-save` 명령 - `runner.command`: 반복 실행용 핵심 명령 - `automation.schedule`: 스케줄 객체 - `automation.cron_line`: 바로 복사 가능한 cron 한 줄 - `automation.system_event_text`: OpenClaw systemEvent 초안 - `automation.openclaw_prompt`: 작업 생성기에 넘기기 좋은 요약 프롬프트 - `assistant_summary.confirmation`: 사용자에게 확인받기 좋은 짧은 문장 `plan` 출력에는 보통 다음이 들어갑니다. - 작업 유형: `monitor / briefing / monitor+briefing` - 해석된 질의 목록 - 일정 종류: `interval / daily / weekly / manual` - cron 힌트 - 추천 저장 대상: `watch / group` - 추천 템플릿 - operator hints - 추천 실행 명령 - 추천 전달 포맷 - cron 예시 - 추천 후속 명령 `plan-save`는 여기에 더해 **실제 DB에 저장되는 watch/group 객체에도 template / schedule / operator_hints / tags / context를 같이 남깁니다.** --- ## 추천 운영 패턴 ### 패턴 1) 새 기사 감시형 요청: - `반도체 뉴스 1시간마다 모니터링해줘` 추천 흐름: ```bash python scripts/naver_news_briefing.py plan "반도체 뉴스 1시간마다 모니터링해줘" python scripts/naver_news_briefing.py plan-save "반도체 뉴스 1시간마다 모니터링해줘" --as watch --name semiconductor-hourly python scripts/naver_news_briefing.py watch-check semiconductor-hourly --json ``` 이 방식은 **신규 기사 JSON을 메신저/웹훅 레이어로 넘기기 좋다**는 장점이 있습니다. ### 패턴 2) 아침 브리핑형 요청: - `매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘` 추천 흐름: ```bash python scripts/naver_news_briefing.py plan-save \ "매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘" \ --as group --name morning-tech --label "아침 브리핑" python scripts/naver_news_briefing.py brief-multi --group morning-tech ``` 이 패턴은 **group에 template를 저장해 두고 cron에서는 group 이름만 호출**하는 방식이 편합니다. ### 패턴 3) 필터 강한 실무형 감시 요청: - `증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘` 추천 흐름: ```bash python scripts/naver_news_briefing.py plan "증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘" python scripts/naver_news_briefing.py plan-save "증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘" --as watch --name samsung-watch python scripts/naver_news_briefing.py watch-check samsung-watch --json ``` --- ## 온보딩 / 최초 사용 가이드 처음 설치한 사용자는 아래 순서로 시작하면 됩니다. 1. 네이버 Search API 자격증명(`client_id`, `client_secret`)을 준비합니다. 2. `setup` 명령으로 최초 입력을 완료합니다. 3. `check-credentials --json`으로 유효성을 확인합니다. 4. 그 다음에만 `search` 또는 `plan`/`plan-save` 흐름으로 넘어갑니다. 가장 안전한 첫 실행 순서: ```bash python scripts/naver_news_briefing.py setup --live-check python scripts/naver_news_briefing.py check-credentials --live-check python scripts/naver_news_briefing.py search "최근 3일 반도체 뉴스 브리핑" ``` 자동화 스크립트나 CI에서는 아래처럼 인자를 직접 넣는 방식도 계속 지원합니다. ```bash python scripts/naver_news_briefing.py setup --client-id YOUR_ID --client-secret YOUR_SECRET python scripts/naver_news_briefing.py check-credentials --json ``` 만약 자격증명이 설정되지 않았다면, 다른 명령을 먼저 시도하기보다 **다시 온보딩(setup)부터 안내하는 쪽이 맞습니다.** ### 자주 보는 초기 오류 자격증명 없이 바로 `search`나 `watch-check`를 실행하면, 이제 CLI는 단순한 실패 문구 대신 아래 방향으로 안내합니다. - 네이버 Search API 자격증명이 아직 설정되지 않았다는 점 - 먼저 `setup`을 실행해야 한다는 점 - 이어서 `check-credentials --json`으로 검증하라는 점 - 그 다음에 실제 검색/브리핑 명령으로 넘어가라는 점 즉, **초기 사용자에게는 에러보다 온보딩 안내가 먼저 보이도록** 맞춰 두었습니다. ### 추천 health check 흐름 실제 API가 정상 응답하는지까지 보고 싶다면 아래 흐름이 가장 직관적입니다. ```bash python scripts/naver_news_briefing.py setup --live-check python scripts/naver_news_briefing.py check-credentials --live-check ``` 원하는 테스트 질의가 있다면: ```bash python scripts/naver_news_briefing.py setup --test-search "삼성전자 뉴스" python scripts/naver_news_briefing.py check-credentials --live-check --query "삼성전자 뉴스" ``` ## 운영자 가이드 ### 1) watch와 group을 구분해서 쓰기 - **watch**: 단일 관심 주제를 새 기사 기준으로 계속 체크할 때 - **group**: 여러 주제를 묶어 반복 브리핑할 때 이 둘을 섞지 말고 역할을 분리하면 운영이 단순해집니다. ### 2) 스케줄은 외부에서, 상태는 이 스킬에서 이 스킬은 다음을 담당합니다. - 질의 정규화 - 상태 저장 - dedupe - 브리핑 렌더링 - 일정 의도 해석 - operator-friendly plan 출력 정확한 실행 시각은 아래 같은 외부 레이어에서 붙이세요. - OpenClaw cron - Windows 작업 스케줄러 - GitHub Actions - 별도 Python worker - PM2 / systemd timer / crontab ### 3) cron 연결 힌트 예를 들어 `plan` 출력이 아래를 주면: - cron 힌트: `0 7 * * *` - 추천 실행 명령: `python scripts/naver_news_briefing.py brief-multi --group morning-tech --template morning-briefing` 운영자는 이를 거의 그대로 옮길 수 있습니다. 예시: ```cron 0 7 * * * cd /path/to/workspace/skills/naver-news-briefing && python scripts/naver_news_briefing.py brief-multi --group morning-tech --template morning-briefing ``` watch 예시: ```cron 0 * * * * cd /path/to/workspace/skills/naver-news-briefing && python scripts/naver_news_briefing.py watch-check semiconductor-hourly --json ``` ### 4) 저장 메타데이터를 같이 남기는 이유 `plan-save`는 단순히 검색어만 저장하지 않습니다. 아래 메타데이터를 같이 남겨 두면 운영자가 나중에 훨씬 덜 헷갈립니다. - `label`: 사람이 읽는 용도 - `tags`: 분류 / 운영 필터링 - `template`: 출력 형식 기본값 - `schedule`: 어떤 cadence로 설계됐는지 - `operator_hints`: 추천 실행 명령 / 전달 방식 / cron 예시 - `context`: 원래 사용자의 요청 문장 ### 5) 가장 안정적인 질의 형식 자연어도 되지만 아래 구조가 가장 예측 가능하게 동작합니다. - 기간 표현 + 핵심 키워드 + 제외어 예: - `최근 7일 반도체 공급망 -광고 -주가` - `오늘 AI 데이터센터 -리포트` ### 6) dedupe 동작 `watch-check`는 `(watch_id, link)` 기준으로 신규 여부를 판정합니다. 같은 기사 링크는 반복 실행해도 다시 알리지 않습니다. --- ## 저장 파일 - `data/config.json`: API 자격증명 및 기본 설정 - `data/watch_state.db`: watch / group / seen-link 상태 - `references/upstream-notes.md`: upstream 설계 메모 --- ## 테스트 ```bash python -m pytest scripts/tests -q ``` --- ## 한계 - 기사 본문 크롤링/본문 요약은 하지 않습니다. - 네이버 Search API의 제목 / 요약 / 링크 / 발행시각 메타데이터 기반으로 동작합니다. - 자연어 일정 파서는 실무형 요청 위주입니다. - 주제 없이 `매일 아침 7시에 브리핑해줘`처럼 일정만 있는 요청은 계획은 일부 만들 수 있어도 저장 가능한 watch/group으로는 제한될 수 있습니다. FILE:references/upstream-notes.md # Upstream notes: `navernews-tabsearch` Use these notes when extending or debugging the skill. ## Reused patterns - `core.query_parser.py` - Preserve the upstream split between: - API query = all positive keywords joined by spaces - tab/db keyword = first positive keyword only - exclude words = `-단어` tokens - Preserve `build_fetch_key(search_query, exclude_words)` normalization. - `core.config_store.py` - Reuse Windows-first credential handling with DPAPI-encrypted `client_secret_enc` and atomic config writes. - `core.workers.ApiWorker` - Reuse request shape: `https://openapi.naver.com/v1/search/news.json` - Use headers `X-Naver-Client-Id`, `X-Naver-Client-Secret` - Use `sort=date`, clean `` tags, prefer `news.naver.com` links when present, filter exclude words on title/description. - Repo persistence style - Keep state on local disk inside the skill (`data/`), use atomic JSON for config and SQLite for watch state / dedupe. ## Intentional simplifications vs upstream GUI app - No PyQt UI, tabs, bookmarks, tray, or backup rotation UI. - No full article database; only persistent watch rules, keyword groups, and seen-link dedupe for cron-style monitoring. - Adapt the upstream tab-search/bookmark mindset into CLI-friendly persistent keyword groups (`group-*`) plus combined briefing templates (`brief-multi`) instead of recreating GUI tabs. - Focus on CLI/stdout output that OpenClaw can relay into chat or scheduled jobs. ## Relevant upstream files - `README.md` - `core/query_parser.py` - `core/config_store.py` - `core/workers.py` - `core/database.py` - `tests/test_query_parser_search_policy.py` FILE:scripts/automation_plans.py from __future__ import annotations import json import re from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Dict, List from query_utils import build_intent, clean_natural_query _MONITOR_WORDS = ["모니터링", "감시", "체크", "알림", "추적", "watch"] _BRIEF_WORDS = ["브리핑", "요약", "정리", "보고"] _GROUP_HINTS = ["묶", "그룹", "여러", "같이", "함께", "이랑", "하고", "랑"] _TEMPLATE_HINTS = { "concise": ["간단", "짧게", "핵심만", "한눈에"], "analyst": ["분석", "인사이트", "시사점", "깊게"], "morning-briefing": ["아침", "출근", "오전 보고", "morning"], "watch-alert": ["알림", "alert", "실시간", "바로"], } _TIME_OF_DAY_HINTS = { "아침": "08:00", "오전": "09:00", "점심": "12:00", "오후": "15:00", "저녁": "18:00", "밤": "21:00", "새벽": "06:00", } _DAYS_OF_WEEK = {"월": "mon", "화": "tue", "수": "wed", "목": "thu", "금": "fri", "토": "sat", "일": "sun"} _DOW_CRON = {"mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6, "sun": 0} @dataclass(frozen=True) class SchedulePlan: kind: str label: str cron: str | None = None interval_minutes: int | None = None time: str | None = None days_of_week: List[str] = field(default_factory=list) @dataclass(frozen=True) class OperatorHints: recommended_runner: str recommended_command: str delivery_format: str storage_target: str cadence_summary: str cron_examples: List[str] = field(default_factory=list) notes: List[str] = field(default_factory=list) @dataclass(frozen=True) class AutomationPlan: raw_request: str action: str query_mode: str queries: List[str] primary_query: str | None intent: Dict[str, Any] | None schedule: SchedulePlan name_hint: str template: str briefing_focus: str watch_intent: str group_reason: str | None rationale: List[str] suggested_commands: List[str] operator_hints: OperatorHints def _normalize_request(raw: str) -> str: return re.sub(r"\s+", " ", str(raw or "").strip()) def _slugify_korean(text: str) -> str: cleaned = re.sub(r"[^0-9A-Za-z가-힣]+", "-", str(text or "").strip().lower()) cleaned = re.sub(r"-+", "-", cleaned).strip("-") return cleaned or "news-plan" def _detect_action(raw: str) -> str: has_monitor = any(word in raw for word in _MONITOR_WORDS) has_brief = any(word in raw for word in _BRIEF_WORDS) if has_monitor and not has_brief: return "monitor" if has_brief and not has_monitor: return "briefing" if has_monitor and has_brief: return "monitor+briefing" return "briefing" def _strip_schedule_and_action_phrases(text: str) -> str: patterns = [ r"\d+\s*시간(?:마다|간격)", r"\d+\s*분(?:마다|간격)", r"매일", r"매주\s*[월화수목금토일]요일?", r"(?:아침|오전|점심|오후|저녁|밤|새벽)\s*\d{1,2}(?::|시)?\s*\d{0,2}분?(?:\s*에)?", r"\d{1,2}(?::|시)\s*\d{0,2}분?(?:\s*에)?", r"실시간", r"수시로", r"계속", r"지속적으로", r"주기적으로", r"모니터링해줘|모니터링 해줘|모니터링", r"감시해줘|감시 해줘|감시", r"체크해줘|체크 해줘|체크", r"알림해줘|알림 해줘", r"브리핑해줘|브리핑 해줘", r"정리해줘|정리 해줘", r"요약해줘|요약 해줘", ] stripped = text for pattern in patterns: stripped = re.sub(pattern, " ", stripped) return _normalize_request(stripped) def _normalize_query_order(query: str) -> str: parts = [part for part in str(query or "").split() if part] positives = [part for part in parts if not part.startswith("-")] negatives = [part for part in parts if part.startswith("-")] return " ".join(positives + negatives) def _extract_queries(raw: str) -> List[str]: text = _strip_schedule_and_action_phrases(_normalize_request(raw)) split_pattern = r"\s*(?:,|/|\+|그리고|및|이랑|와|과|랑|하고)\s*" candidates = [part.strip() for part in re.split(split_pattern, text) if part.strip()] queries: List[str] = [] for candidate in candidates: cleaned = _normalize_query_order(clean_natural_query(candidate)) if cleaned: queries.append(cleaned) deduped: List[str] = [] seen = set() for query in queries: key = query.lower() if key in seen: continue seen.add(key) deduped.append(query) return deduped def _detect_explicit_time(text: str) -> str | None: prefixed = re.search(r"(아침|오전|오후|저녁|밤|새벽)\s*(\d{1,2})(?:[:시]\s*(\d{1,2}))?분?", text) if prefixed: prefix = prefixed.group(1) hour = int(prefixed.group(2)) minute = int(prefixed.group(3) or 0) if prefix in {"오후", "저녁", "밤"} and hour < 12: hour += 12 if prefix == "새벽" and hour == 12: hour = 0 if prefix == "오전" and hour == 12: hour = 0 return f"{hour:02d}:{minute:02d}" plain = re.search(r"(\d{1,2})(?:[:시]\s*(\d{1,2}))?분?", text) if plain: return f"{int(plain.group(1)):02d}:{int(plain.group(2) or 0):02d}" return None def _detect_schedule(raw: str) -> SchedulePlan: text = _normalize_request(raw) if re.search(r"실시간|수시로|계속|지속적으로", text): return SchedulePlan(kind="interval", label="15분마다", cron="*/15 * * * *", interval_minutes=15) every_n_hours = re.search(r"(\d+)\s*시간(?:마다|간격)", text) if every_n_hours: hours = max(1, int(every_n_hours.group(1))) minutes = hours * 60 cron = f"0 */{hours} * * *" if minutes >= 60 and minutes % 60 == 0 else f"*/{minutes} * * * *" return SchedulePlan(kind="interval", label=f"{hours}시간마다", cron=cron, interval_minutes=minutes) every_n_minutes = re.search(r"(\d+)\s*분(?:마다|간격)", text) if every_n_minutes: minutes = max(1, int(every_n_minutes.group(1))) return SchedulePlan(kind="interval", label=f"{minutes}분마다", cron=f"*/{minutes} * * * *", interval_minutes=minutes) weekly = re.search(r"매주\s*([월화수목금토일])요일?", text) if weekly: day = _DAYS_OF_WEEK[weekly.group(1)] time = _detect_explicit_time(text) or next((v for k, v in _TIME_OF_DAY_HINTS.items() if k in text), "08:00") hh, mm = time.split(":") return SchedulePlan(kind="weekly", label=f"매주 {weekly.group(1)}요일 {time}", cron=f"{int(mm)} {int(hh)} * * {_DOW_CRON[day]}", time=time, days_of_week=[day]) if "매일" in text: time = _detect_explicit_time(text) or next((v for k, v in _TIME_OF_DAY_HINTS.items() if k in text), "08:00") hh, mm = time.split(":") return SchedulePlan(kind="daily", label=f"매일 {time}", cron=f"{int(mm)} {int(hh)} * * *", time=time) return SchedulePlan(kind="manual", label="수동 실행") def _detect_template(raw: str, action: str, schedule: SchedulePlan) -> str: for template, words in _TEMPLATE_HINTS.items(): if any(word in raw for word in words): return template if action == "monitor": return "watch-alert" if schedule.kind == "daily": return "morning-briefing" return "concise" def _detect_briefing_focus(raw: str, template: str) -> str: if template == "morning-briefing": return "아침 브리핑" if any(word in raw for word in ["핵심만", "간단", "짧게"]): return "핵심 요약" if any(word in raw for word in ["분석", "시사점", "인사이트"]): return "분석형 정리" return "일반 브리핑" def _detect_watch_intent(raw: str, action: str) -> str: if "monitor" not in action: return "none" if any(word in raw for word in ["계속", "실시간", "수시로", "바로"]): return "continuous" if any(word in raw for word in ["모니터링", "감시", "체크", "알림"]): return "scheduled" return "manual" def _suggest_name(queries: List[str], action: str) -> str: base = queries[0] if queries else action suffix = "watch" if "monitor" in action else "brief" return _slugify_korean(f"{base}-{suffix}") def _build_commands(plan: AutomationPlan) -> List[str]: commands: List[str] = [] if plan.query_mode == "group": quoted_queries = " ".join(f'"{query}"' for query in plan.queries) commands.append( f'python scripts/naver_news_briefing.py group-add {plan.name_hint} {quoted_queries} --label "자동 생성 그룹" --context "{plan.raw_request}" --template {plan.template}' ) commands.append(f"python scripts/naver_news_briefing.py brief-multi --group {plan.name_hint} --template {plan.template}") commands.append(f'python scripts/naver_news_briefing.py plan-save "{plan.raw_request}" --as group --name {plan.name_hint}') else: query = plan.primary_query or "" if "monitor" in plan.action: commands.append(f'python scripts/naver_news_briefing.py watch-add {plan.name_hint} "{query}" --template {plan.template}') commands.append(f"python scripts/naver_news_briefing.py watch-check {plan.name_hint} --json") if "briefing" in plan.action: commands.append(f'python scripts/naver_news_briefing.py search "{query}"') commands.append(f'python scripts/naver_news_briefing.py plan-save "{plan.raw_request}" --as watch --name {plan.name_hint}') return commands def _build_operator_hints(plan: AutomationPlan) -> OperatorHints: storage_target = "group" if plan.query_mode == "group" else "watch" recommended_runner = "cron" if plan.schedule.kind != "manual" else "manual" if storage_target == "group": recommended_command = f"python scripts/naver_news_briefing.py brief-multi --group {plan.name_hint} --template {plan.template}" delivery_format = "chat-briefing" else: recommended_command = f"python scripts/naver_news_briefing.py watch-check {plan.name_hint} --json" if "monitor" in plan.action else f'python scripts/naver_news_briefing.py search "{plan.primary_query or ""}"' delivery_format = "watch-json" if "monitor" in plan.action else "chat-briefing" notes = [ "스케줄 실행은 이 스킬이 아니라 외부 cron/작업 스케줄러/OpenClaw cron에서 연결하는 방식이 가장 안정적입니다.", "watch는 새 기사 감지용, group은 반복 브리핑용으로 두면 운영이 단순합니다.", ] cron_examples: List[str] = [] if plan.schedule.cron: cron_examples.append(f'{plan.schedule.cron} {recommended_command}') cron_examples.append(f'{plan.schedule.cron} # 위 명령 실행 후 stdout/json을 텔레그램·디스코드 전송') return OperatorHints( recommended_runner=recommended_runner, recommended_command=recommended_command, delivery_format=delivery_format, storage_target=storage_target, cadence_summary=plan.schedule.label, cron_examples=cron_examples, notes=notes, ) def parse_automation_request(raw: str) -> AutomationPlan: request = _normalize_request(raw) action = _detect_action(request) schedule = _detect_schedule(request) queries = _extract_queries(request) query_mode = "group" if len(queries) > 1 or any(hint in request for hint in _GROUP_HINTS) else "single" primary_query = queries[0] if queries else None intent = None rationale: List[str] = [] if primary_query: built = build_intent(primary_query) intent = asdict(built) rationale.append(f"핵심 검색어를 '{built.search_query}'로 정규화했습니다.") if built.exclude_words: rationale.append("제외어를 유지해 watch/search 명령으로 바로 연결할 수 있게 했습니다.") else: rationale.append("주제 키워드가 없어 저장 가능한 watch/search 명령은 만들지 않았습니다.") if schedule.kind != "manual": rationale.append(f"일정 표현을 '{schedule.label}' 구조로 해석했습니다.") group_reason = None if query_mode == "group": group_reason = "여러 주제를 한 번에 반복 브리핑하기 좋은 요청입니다." rationale.append("여러 주제를 감지해 그룹 기반 브리핑/자동화로 분류했습니다.") template = _detect_template(request, action, schedule) plan = AutomationPlan( raw_request=request, action=action, query_mode=query_mode, queries=queries, primary_query=primary_query, intent=intent, schedule=schedule, name_hint=_suggest_name(queries, action), template=template, briefing_focus=_detect_briefing_focus(request, template), watch_intent=_detect_watch_intent(request, action), group_reason=group_reason, rationale=rationale, suggested_commands=[], operator_hints=OperatorHints( recommended_runner="manual", recommended_command="", delivery_format="text", storage_target="watch" if query_mode == "single" else "group", cadence_summary=schedule.label, ), ) commands = _build_commands(plan) hints = _build_operator_hints(plan) return AutomationPlan(**{**asdict(plan), "suggested_commands": commands, "operator_hints": hints, "schedule": plan.schedule}) def render_plan_text(plan: AutomationPlan) -> str: lines = ["## 뉴스 자동화 계획", f"- 요청: {plan.raw_request}", f"- 작업 유형: {plan.action}", f"- 일정: {plan.schedule.label}"] if plan.queries: lines.append("- 해석된 질의: " + ", ".join(plan.queries)) lines.append(f"- 저장 대상 추천: {plan.operator_hints.storage_target}") if plan.name_hint: lines.append(f"- 저장 이름 제안: {plan.name_hint}") lines.append(f"- 추천 템플릿: {plan.template} ({plan.briefing_focus})") lines.append(f"- watch 의도: {plan.watch_intent}") if plan.schedule.cron: lines.append(f"- cron 힌트: {plan.schedule.cron}") if plan.group_reason: lines.append(f"- 그룹 판단: {plan.group_reason}") if plan.rationale: lines.append("- 해석 근거:") lines.extend(f" - {item}" for item in plan.rationale) lines.append("- 운영 힌트:") lines.append(f" - 실행 방식: {plan.operator_hints.recommended_runner}") lines.append(f" - 추천 실행 명령: {plan.operator_hints.recommended_command}") for note in plan.operator_hints.notes: lines.append(f" - {note}") if plan.operator_hints.cron_examples: lines.append("- cron 예시:") lines.extend(f" - {item}" for item in plan.operator_hints.cron_examples) if plan.suggested_commands: lines.append("- 추천 명령:") lines.extend(f" - {cmd}" for cmd in plan.suggested_commands) return "\n".join(lines) def plan_to_dict(plan: AutomationPlan) -> Dict[str, Any]: payload = asdict(plan) payload["schedule"] = asdict(plan.schedule) payload["operator_hints"] = asdict(plan.operator_hints) return payload def build_integration_bundle(raw: str, *, skill_dir: str | Path | None = None, assistant_channel: str = "telegram") -> Dict[str, Any]: plan = parse_automation_request(raw) skill_root = Path(skill_dir).resolve() if skill_dir else Path(__file__).resolve().parents[1] cli_rel = str(skill_root / "scripts" / "naver_news_briefing.py") save_as = "group" if plan.query_mode == "group" else "watch" save_command = f'python scripts/naver_news_briefing.py plan-save "{plan.raw_request}" --as {save_as} --name {plan.name_hint}' run_command = plan.operator_hints.recommended_command shell_run = f'cd "{skill_root}" && {run_command}' shell_save = f'cd "{skill_root}" && {save_command}' schedule_payload = { "kind": plan.schedule.kind, "label": plan.schedule.label, "cron": plan.schedule.cron, "time": plan.schedule.time, "interval_minutes": plan.schedule.interval_minutes, "days_of_week": plan.schedule.days_of_week, } confirmation = f"'{plan.raw_request}' 요청을 {plan.schedule.label} {plan.operator_hints.storage_target} 자동화로 해석했습니다. 저장 이름은 '{plan.name_hint}'를 추천하고, 저장 뒤에는 '{run_command}'를 스케줄러에 연결하면 됩니다." cadence_phrase = plan.schedule.label if plan.schedule.label.endswith("마다") else f"{plan.schedule.label}마다" system_event_text = ( f"{cadence_phrase} {assistant_channel} 채널에 네이버 뉴스 자동화를 실행하세요. " f"먼저 {save_command} 로 상태를 저장하고, 이후 {run_command} 결과를 전달합니다." if plan.schedule.kind != "manual" else f"필요할 때 {run_command} 를 수동 실행하는 네이버 뉴스 브리핑 요청입니다." ) openclaw_prompt = ( "다음 뉴스 자동화 계획을 기준으로 cron/작업을 생성하세요.\n" f"- 사용자 요청: {plan.raw_request}\n" f"- 저장 명령: {save_command}\n" f"- 실행 명령: {run_command}\n" f"- 일정: {json.dumps(schedule_payload, ensure_ascii=False)}\n" f"- 전달 포맷: {plan.operator_hints.delivery_format}\n" f"- 확인 문구: {confirmation}" ) bundle = { "plan": plan_to_dict(plan), "storage": { "target": save_as, "name": plan.name_hint, "save_command": save_command, "shell_save_command": shell_save, }, "runner": { "command": run_command, "shell_command": shell_run, "delivery_format": plan.operator_hints.delivery_format, "assistant_channel": assistant_channel, }, "automation": { "schedule": schedule_payload, "cron_line": f"{plan.schedule.cron} {shell_run}" if plan.schedule.cron else None, "system_event_text": system_event_text, "openclaw_prompt": openclaw_prompt, }, "assistant_summary": { "confirmation": confirmation, "user_summary": confirmation, "next_step": "저장 명령을 한 번 실행한 뒤 cron/OpenClaw 작업에서 실행 명령을 연결하세요.", }, "artifacts": { "skill_dir": str(skill_root), "cli_path": cli_rel, }, } return bundle def render_integration_bundle_text(bundle: Dict[str, Any]) -> str: plan = bundle["plan"] storage = bundle["storage"] runner = bundle["runner"] automation = bundle["automation"] assistant_summary = bundle["assistant_summary"] lines = [ "## OpenClaw 연동 번들", f"- 요청: {plan['raw_request']}", f"- 저장 대상: {storage['target']} ({storage['name']})", f"- 일정: {automation['schedule']['label']}", f"- 저장 명령: {storage['save_command']}", f"- 실행 명령: {runner['command']}", ] if automation.get("cron_line"): lines.append(f"- cron 한 줄 예시: {automation['cron_line']}") lines.append("- OpenClaw systemEvent 제안:") lines.append(f" {automation['system_event_text']}") lines.append("- OpenClaw 작업 생성용 프롬프트:") for line in str(automation["openclaw_prompt"]).splitlines(): lines.append(f" {line}") lines.append("- 사용자 확인 문구:") lines.append(f" {assistant_summary['confirmation']}") return "\n".join(lines) FILE:scripts/briefing_templates.py from __future__ import annotations import json from collections import Counter from typing import Any, Dict, List TEMPLATES = { "concise": "핵심만 빠르게 묶는 짧은 브리핑", "analyst": "질문별 핵심 흐름과 시사점을 조금 더 분석적으로 정리", "morning-briefing": "아침 보고용으로 전체 동향과 체크포인트를 정리", "watch-alert": "감시/알림 용도로 신규 기사 중심으로 짧게 정리", } def supported_templates() -> List[str]: return list(TEMPLATES.keys()) def build_combined_payload(entries: List[Dict[str, Any]], *, template: str, source_groups: List[Dict[str, Any]] | None = None) -> Dict[str, Any]: publishers = Counter() total_items = 0 total_filtered = 0 total_too_old = 0 for entry in entries: result = entry["result"] total_items += len(result.get("items", [])) total_filtered += int(result.get("filtered_out", 0) or 0) total_too_old += int(result.get("too_old", 0) or 0) for item in result.get("items", []): publisher = str(item.get("publisher", "") or "").strip() if publisher: publishers[publisher] += 1 top_publishers = [{"publisher": name, "count": count} for name, count in publishers.most_common(5)] return { "template": template, "entry_count": len(entries), "item_count": total_items, "filtered_out": total_filtered, "too_old": total_too_old, "top_publishers": top_publishers, "groups": source_groups or [], "entries": entries, } def _header(payload: Dict[str, Any]) -> List[str]: lines = [f"네이버 뉴스 멀티 브리핑 · {payload['template']}"] lines.append(f"- 쿼리 수: {payload['entry_count']} / 기사 수: {payload['item_count']}") if payload.get("filtered_out"): lines.append(f"- 제외어 필터링: {payload['filtered_out']}건") if payload.get("too_old"): lines.append(f"- 기간 제외: {payload['too_old']}건") if payload.get("groups"): group_names = ", ".join(group["name"] for group in payload["groups"]) lines.append(f"- 그룹: {group_names}") top_publishers = payload.get("top_publishers") or [] if top_publishers: lines.append("- 많이 보인 출처: " + ", ".join(f"{item['publisher']} {item['count']}건" for item in top_publishers[:3])) return lines def _entry_title(entry: Dict[str, Any]) -> str: title_bits = [] if entry.get("group_name"): title_bits.append(f"[{entry['group_name']}]") title_bits.append(entry["query"]) if entry.get("label"): title_bits.append(f"({entry['label']})") return " ".join(title_bits) def _entry_items(entry: Dict[str, Any], *, limit: int) -> List[str]: result = entry["result"] items = result.get("items", [])[:limit] if not items: return ["- 표시할 기사가 없습니다."] lines: List[str] = [] for item in items: title = str(item.get("title", "") or "").strip() publisher = str(item.get("publisher", "정보 없음") or "정보 없음") lines.append(f"- [{publisher}] {title}") if item.get("link"): lines.append(f" 링크: {item['link']}") return lines def render_combined_text(payload: Dict[str, Any]) -> str: template = payload["template"] lines = _header(payload) lines.append("") if template == "concise": for entry in payload["entries"]: lines.append(f"## {_entry_title(entry)}") lines.extend(_entry_items(entry, limit=2)) lines.append("") elif template == "analyst": lines.append("관찰 포인트") for entry in payload["entries"]: result = entry["result"] lines.append(f"## {_entry_title(entry)}") lines.append(f"- 노출 기사: {result.get('displayed', 0)} / 전체 결과: {result.get('total', 0)}") if entry.get("context"): lines.append(f"- 맥락: {entry['context']}") lines.extend(_entry_items(entry, limit=3)) lines.append("") elif template == "morning-briefing": lines.append("오늘 체크할 흐름") for entry in payload["entries"]: result = entry["result"] headline = result.get("items", [{}])[0].get("title") if result.get("items") else "기사 없음" lines.append(f"- {_entry_title(entry)}: {headline}") lines.append("") lines.append("세부 기사") for entry in payload["entries"]: lines.append(f"## {_entry_title(entry)}") lines.extend(_entry_items(entry, limit=2)) lines.append("") elif template == "watch-alert": alert_entries = [entry for entry in payload["entries"] if entry["result"].get("displayed", 0)] if not alert_entries: lines.append("- 신규/표시 가능한 기사가 없습니다.") for entry in alert_entries: lines.append(f"## ALERT · {_entry_title(entry)}") lines.extend(_entry_items(entry, limit=2)) lines.append("") else: raise ValueError(f"지원하지 않는 템플릿입니다: {template}") return "\n".join(line for line in lines).strip() def render_combined_json(payload: Dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False, indent=2) FILE:scripts/config_store.py from __future__ import annotations import base64 import ctypes import json import os import sys import tempfile from typing import Any, Dict, Mapping, Tuple from _paths import CONFIG_PATH, ensure_data_dir DEFAULT_CONFIG: Dict[str, Any] = { "naver_api": { "client_id": "", "client_secret": "", "client_secret_enc": "", "client_secret_storage": "plain", "timeout": 15, } } def _is_windows_platform() -> bool: return sys.platform == "win32" def _normalize_secret_storage(value: Any) -> str: return "dpapi" if str(value or "").strip().lower() == "dpapi" else "plain" def _dpapi_encrypt_text(text: str) -> str: if not _is_windows_platform() or not text: return "" from ctypes import wintypes class DATA_BLOB(ctypes.Structure): _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte))] source = text.encode("utf-8") source_buffer = (ctypes.c_byte * len(source)).from_buffer_copy(source) in_blob = DATA_BLOB(len(source), ctypes.cast(source_buffer, ctypes.POINTER(ctypes.c_byte))) out_blob = DATA_BLOB() crypt32 = ctypes.windll.crypt32 kernel32 = ctypes.windll.kernel32 ok = crypt32.CryptProtectData(ctypes.byref(in_blob), None, None, None, None, 0, ctypes.byref(out_blob)) if not ok: return "" try: encrypted = ctypes.string_at(out_blob.pbData, out_blob.cbData) return base64.b64encode(encrypted).decode("ascii") finally: if out_blob.pbData: kernel32.LocalFree(out_blob.pbData) def _dpapi_decrypt_text(payload: str) -> str: if not _is_windows_platform() or not payload: return "" from ctypes import wintypes raw = base64.b64decode(payload) class DATA_BLOB(ctypes.Structure): _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte))] source_buffer = (ctypes.c_byte * len(raw)).from_buffer_copy(raw) in_blob = DATA_BLOB(len(raw), ctypes.cast(source_buffer, ctypes.POINTER(ctypes.c_byte))) out_blob = DATA_BLOB() crypt32 = ctypes.windll.crypt32 kernel32 = ctypes.windll.kernel32 ok = crypt32.CryptUnprotectData(ctypes.byref(in_blob), None, None, None, None, 0, ctypes.byref(out_blob)) if not ok: return "" try: return ctypes.string_at(out_blob.pbData, out_blob.cbData).decode("utf-8") finally: if out_blob.pbData: kernel32.LocalFree(out_blob.pbData) def encode_client_secret_for_storage(client_secret: str) -> Dict[str, str]: plain = str(client_secret or "").strip() if not plain: return {"client_secret": "", "client_secret_enc": "", "client_secret_storage": "plain"} if _is_windows_platform(): encrypted = _dpapi_encrypt_text(plain) if encrypted: return {"client_secret": "", "client_secret_enc": encrypted, "client_secret_storage": "dpapi"} return {"client_secret": plain, "client_secret_enc": "", "client_secret_storage": "plain"} def resolve_client_secret_for_runtime(settings: Mapping[str, Any]) -> Tuple[str, bool]: plain = str(settings.get("client_secret", "") or "") encrypted = str(settings.get("client_secret_enc", "") or "") storage = _normalize_secret_storage(settings.get("client_secret_storage", "plain")) if _is_windows_platform() and encrypted and storage == "dpapi": decrypted = _dpapi_decrypt_text(encrypted) if decrypted: return decrypted, bool(plain) return plain, bool(_is_windows_platform() and plain) def _write_text_atomic(path: str, text: str) -> None: directory = os.path.dirname(os.path.abspath(path)) or "." os.makedirs(directory, exist_ok=True) fd, tmp_path = tempfile.mkstemp(prefix=".config_", suffix=".tmp", dir=directory) try: with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f: f.write(text) if not text.endswith("\n"): f.write("\n") f.flush() os.fsync(f.fileno()) os.replace(tmp_path, path) finally: if os.path.exists(tmp_path): os.remove(tmp_path) def load_config() -> Dict[str, Any]: ensure_data_dir() if not CONFIG_PATH.exists(): return json.loads(json.dumps(DEFAULT_CONFIG)) raw = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) merged = json.loads(json.dumps(DEFAULT_CONFIG)) if isinstance(raw, dict): merged.update(raw) if isinstance(raw.get("naver_api"), dict): merged["naver_api"].update(raw["naver_api"]) return merged def save_config(config: Dict[str, Any]) -> None: ensure_data_dir() _write_text_atomic(str(CONFIG_PATH), json.dumps(config, indent=2, ensure_ascii=False)) def set_credentials(client_id: str, client_secret: str, timeout: int = 15) -> Dict[str, Any]: config = load_config() encoded = encode_client_secret_for_storage(client_secret) config["naver_api"].update( { "client_id": str(client_id or "").strip(), "client_secret": encoded["client_secret"], "client_secret_enc": encoded["client_secret_enc"], "client_secret_storage": encoded["client_secret_storage"], "timeout": max(5, min(60, int(timeout))), } ) save_config(config) return config def get_runtime_credentials() -> Tuple[str, str, int, Dict[str, Any]]: config = load_config() settings = config["naver_api"] client_secret, needs_migration = resolve_client_secret_for_runtime(settings) if needs_migration and client_secret: updated = set_credentials(str(settings.get("client_id", "")), client_secret, int(settings.get("timeout", 15))) settings = updated["naver_api"] client_secret, _ = resolve_client_secret_for_runtime(settings) return str(settings.get("client_id", "")).strip(), client_secret.strip(), int(settings.get("timeout", 15)), config FILE:scripts/group_store.py from __future__ import annotations import json import sqlite3 from contextlib import contextmanager from datetime import datetime from typing import Any, Dict, Iterable, List from _paths import DB_PATH, ensure_data_dir SCHEMA = """ CREATE TABLE IF NOT EXISTS keyword_groups ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, label TEXT, tags_json TEXT NOT NULL DEFAULT '[]', context TEXT, template TEXT, schedule_json TEXT NOT NULL DEFAULT '{}', operator_hint_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS group_queries ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER NOT NULL, position INTEGER NOT NULL, raw_query TEXT NOT NULL, FOREIGN KEY (group_id) REFERENCES keyword_groups(id) ON DELETE CASCADE, UNIQUE(group_id, position) ); """ @contextmanager def connect(): ensure_data_dir() conn = sqlite3.connect(DB_PATH) try: conn.execute("PRAGMA foreign_keys = ON") conn.executescript(SCHEMA) columns = {row[1] for row in conn.execute("PRAGMA table_info(keyword_groups)").fetchall()} migrations = { "template": "ALTER TABLE keyword_groups ADD COLUMN template TEXT", "schedule_json": "ALTER TABLE keyword_groups ADD COLUMN schedule_json TEXT NOT NULL DEFAULT '{}'", "operator_hint_json": "ALTER TABLE keyword_groups ADD COLUMN operator_hint_json TEXT NOT NULL DEFAULT '{}'", } for column, sql in migrations.items(): if column not in columns: conn.execute(sql) yield conn conn.commit() finally: conn.close() def _normalize_tags(tags: Iterable[str] | None) -> List[str]: seen = set() normalized: List[str] = [] for tag in tags or []: value = str(tag or "").strip() if not value or value in seen: continue seen.add(value) normalized.append(value) return normalized def _serialize_group_row(row: sqlite3.Row | tuple[Any, ...], queries: List[str]) -> Dict[str, Any]: return { "id": row[0], "name": row[1], "label": row[2], "tags": json.loads(row[3] or "[]"), "context": row[4], "template": row[5], "schedule": json.loads(row[6] or "{}"), "operator_hints": json.loads(row[7] or "{}"), "created_at": row[8], "updated_at": row[9], "queries": queries, "query_count": len(queries), } def _fetch_queries(conn: sqlite3.Connection, group_id: int) -> List[str]: rows = conn.execute("SELECT raw_query FROM group_queries WHERE group_id = ? ORDER BY position, id", (group_id,)).fetchall() return [row[0] for row in rows] def list_groups() -> List[Dict[str, Any]]: with connect() as conn: rows = conn.execute("SELECT id, name, label, tags_json, context, template, schedule_json, operator_hint_json, created_at, updated_at FROM keyword_groups ORDER BY name").fetchall() return [_serialize_group_row(row, _fetch_queries(conn, row[0])) for row in rows] def get_group(name_or_id: str | int) -> Dict[str, Any]: field = "id" if isinstance(name_or_id, int) or str(name_or_id).isdigit() else "name" with connect() as conn: row = conn.execute( f"SELECT id, name, label, tags_json, context, template, schedule_json, operator_hint_json, created_at, updated_at FROM keyword_groups WHERE {field} = ?", (int(name_or_id) if field == "id" else name_or_id,), ).fetchone() if not row: raise KeyError(f"keyword group not found: {name_or_id}") return _serialize_group_row(row, _fetch_queries(conn, row[0])) def create_group(*, name: str, queries: List[str], label: str | None = None, tags: Iterable[str] | None = None, context: str | None = None, template: str | None = None, schedule: Dict[str, Any] | None = None, operator_hints: Dict[str, Any] | None = None) -> Dict[str, Any]: clean_queries = [str(query or "").strip() for query in queries if str(query or "").strip()] if not clean_queries: raise ValueError("키워드 그룹에는 최소 1개 이상의 쿼리가 필요합니다.") now = datetime.now().isoformat(timespec="seconds") tag_values = _normalize_tags(tags) with connect() as conn: cur = conn.execute( "INSERT INTO keyword_groups(name, label, tags_json, context, template, schedule_json, operator_hint_json, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", (name, label, json.dumps(tag_values, ensure_ascii=False), context, template, json.dumps(schedule or {}, ensure_ascii=False), json.dumps(operator_hints or {}, ensure_ascii=False), now, now), ) group_id = int(cur.lastrowid) for idx, query in enumerate(clean_queries, start=1): conn.execute("INSERT INTO group_queries(group_id, position, raw_query) VALUES (?, ?, ?)", (group_id, idx, query)) return get_group(name) def update_group(name_or_id: str | int, *, label: str | None = None, context: str | None = None, tags: Iterable[str] | None | object = None, template: str | None = None, schedule: Dict[str, Any] | None = None, operator_hints: Dict[str, Any] | None = None, replace_queries: List[str] | None = None, add_queries: List[str] | None = None, remove_queries: List[str] | None = None) -> Dict[str, Any]: group = get_group(name_or_id) group_id = group["id"] new_label = group["label"] if label is None else label new_context = group["context"] if context is None else context new_template = group["template"] if template is None else template new_schedule = group["schedule"] if schedule is None else schedule new_operator_hints = group["operator_hints"] if operator_hints is None else operator_hints new_tags = group["tags"] if tags is None else _normalize_tags(tags) # type: ignore[arg-type] queries = list(group["queries"]) if replace_queries is not None: queries = [str(query or "").strip() for query in replace_queries if str(query or "").strip()] else: for query in add_queries or []: cleaned = str(query or "").strip() if cleaned and cleaned not in queries: queries.append(cleaned) for query in remove_queries or []: cleaned = str(query or "").strip() queries = [item for item in queries if item != cleaned] if not queries: raise ValueError("키워드 그룹에는 최소 1개 이상의 쿼리가 남아 있어야 합니다.") now = datetime.now().isoformat(timespec="seconds") with connect() as conn: conn.execute( "UPDATE keyword_groups SET label = ?, tags_json = ?, context = ?, template = ?, schedule_json = ?, operator_hint_json = ?, updated_at = ? WHERE id = ?", (new_label, json.dumps(new_tags, ensure_ascii=False), new_context, new_template, json.dumps(new_schedule, ensure_ascii=False), json.dumps(new_operator_hints, ensure_ascii=False), now, group_id), ) conn.execute("DELETE FROM group_queries WHERE group_id = ?", (group_id,)) for idx, query in enumerate(queries, start=1): conn.execute("INSERT INTO group_queries(group_id, position, raw_query) VALUES (?, ?, ?)", (group_id, idx, query)) return get_group(group_id) def remove_group(name_or_id: str | int) -> int: field = "id" if isinstance(name_or_id, int) or str(name_or_id).isdigit() else "name" with connect() as conn: cur = conn.execute(f"DELETE FROM keyword_groups WHERE {field} = ?", (int(name_or_id) if field == "id" else name_or_id,)) return int(cur.rowcount or 0) FILE:scripts/naver_api.py from __future__ import annotations import html import json import re import urllib.parse from dataclasses import dataclass from datetime import datetime, timezone from email.utils import parsedate_to_datetime from typing import Any, Dict, List, Protocol import requests RE_BOLD_TAGS = re.compile(r"") API_URL = "https://openapi.naver.com/v1/search/news.json" class ResponseLike(Protocol): status_code: int text: str def json(self) -> Dict[str, Any]: ... class SessionLike(Protocol): def get(self, url: str, *, headers: Dict[str, str], params: Dict[str, Any], timeout: int) -> ResponseLike: ... @dataclass class NewsItem: title: str description: str link: str original_link: str publisher: str pub_date: str pub_date_iso: str | None def to_dict(self) -> Dict[str, Any]: return self.__dict__.copy() def parse_pub_date(value: str) -> str | None: text = str(value or "").strip() if not text: return None try: dt = parsedate_to_datetime(text) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone().isoformat(timespec="seconds") except Exception: return None def clean_item(item: Dict[str, Any]) -> NewsItem: title = html.unescape(RE_BOLD_TAGS.sub("", str(item.get("title", "")))) desc = html.unescape(RE_BOLD_TAGS.sub("", str(item.get("description", "")))) naver_link = str(item.get("link", "") or "") original_link = str(item.get("originallink", "") or "") if "news.naver.com" in naver_link: final_link = naver_link elif "news.naver.com" in original_link: final_link = original_link else: final_link = naver_link or original_link publisher = "정보 없음" if original_link: publisher = urllib.parse.urlparse(original_link).netloc.replace("www.", "") or publisher elif final_link: publisher = "네이버뉴스" if "news.naver.com" in final_link else urllib.parse.urlparse(final_link).netloc.replace("www.", "") return NewsItem( title=title, description=desc, link=final_link, original_link=original_link, publisher=publisher or "정보 없음", pub_date=str(item.get("pubDate", "") or ""), pub_date_iso=parse_pub_date(str(item.get("pubDate", "") or "")), ) def fetch_news( *, client_id: str, client_secret: str, search_query: str, exclude_words: List[str], limit: int = 10, days: int | None = None, timeout: int = 15, session: SessionLike | None = None, ) -> Dict[str, Any]: if not search_query.strip(): raise ValueError("검색어가 비어 있습니다.") if not client_id.strip() or not client_secret.strip(): raise ValueError("네이버 API 자격증명이 설정되지 않았습니다.") active_session = session or requests.Session() resp = active_session.get( API_URL, headers={ "X-Naver-Client-Id": client_id.strip(), "X-Naver-Client-Secret": client_secret.strip(), }, params={"query": search_query, "display": min(max(limit, 10), 100), "start": 1, "sort": "date"}, timeout=timeout, ) if resp.status_code != 200: try: payload = resp.json() msg = f"API 오류 {resp.status_code} ({payload.get('errorCode', '')}): {payload.get('errorMessage', '알 수 없는 오류')}" except Exception: msg = f"API 오류 {resp.status_code}: {getattr(resp, 'text', '')[:200]}" raise RuntimeError(msg) data = resp.json() exclude_words_lc = [w.lower() for w in exclude_words if w] cutoff = None if days: cutoff = datetime.now().astimezone().timestamp() - (days * 86400) items: List[NewsItem] = [] filtered_out = 0 too_old = 0 for raw in data.get("items", []): item = clean_item(raw) text_blob = f"{item.title}\n{item.description}".lower() if exclude_words_lc and any(ex in text_blob for ex in exclude_words_lc): filtered_out += 1 continue if cutoff is not None and item.pub_date_iso: try: if datetime.fromisoformat(item.pub_date_iso).timestamp() < cutoff: too_old += 1 continue except Exception: pass items.append(item) if len(items) >= limit: break return { "query": search_query, "exclude_words": exclude_words, "days": days, "total": int(data.get("total", 0) or 0), "displayed": len(items), "filtered_out": filtered_out, "too_old": too_old, "items": [item.to_dict() for item in items], "raw_meta": {"lastBuildDate": data.get("lastBuildDate", "")}, } FILE:scripts/naver_news_briefing.py from __future__ import annotations import argparse import getpass import html import json import re import sys from datetime import datetime from pathlib import Path from typing import Any, Dict, List from automation_plans import ( build_integration_bundle, parse_automation_request, plan_to_dict, render_integration_bundle_text, render_plan_text, ) from briefing_templates import build_combined_payload, render_combined_json, render_combined_text, supported_templates from config_store import get_runtime_credentials, set_credentials from group_store import create_group, get_group, list_groups, remove_group, update_group from naver_api import fetch_news from query_utils import build_intent from watch_store import add_rule, get_rule, list_rules, mark_seen, remove_rule MISSING_CREDENTIALS_ERROR = "네이버 API 자격증명이 설정되지 않았습니다." DEFAULT_TEST_QUERY = "최근 1일 반도체 뉴스" ANNOUNCE_PREVIEW_LIMIT = 10 def _strip_html(value: str) -> str: text = html.unescape(str(value or "")) text = re.sub(r"<[^>]+>", "", text) return " ".join(text.split()).strip() def _brief_lines(result: Dict[str, Any], *, title: str | None = None) -> List[str]: lines: List[str] = [] heading = title or f"네이버 뉴스 브리핑: {result['query']}" suffix = [] if result.get("exclude_words"): suffix.append("제외=" + ", ".join(result["exclude_words"])) if result.get("days"): suffix.append(f"최근 {result['days']}일") meta = f" ({'; '.join(suffix)})" if suffix else "" lines.append(heading + meta) lines.append(f"- 총 검색 결과: {result.get('total', 0)}") lines.append(f"- 출력 기사 수: {result.get('displayed', 0)}") if result.get("filtered_out"): lines.append(f"- 제외어로 걸러진 수: {result['filtered_out']}") if result.get("too_old"): lines.append(f"- 기간 조건으로 제외된 수: {result['too_old']}") items = result.get("items", []) if not items: lines.append("- 새로 보여줄 기사가 없습니다.") return lines for idx, item in enumerate(items, start=1): publisher = item.get("publisher", "정보 없음") pub_date = item.get("pub_date", "") lines.append(f"{idx}. [{publisher}] {item.get('title', '').strip()}") if item.get("description"): lines.append(f" - {item['description'].strip()[:180]}") if pub_date: lines.append(f" - 발행: {pub_date}") if item.get("link"): lines.append(f" - 링크: {item['link']}") return lines def _print_payload(payload: Any, *, as_json: bool, render_text=None) -> None: if as_json: print(json.dumps(payload, ensure_ascii=False, indent=2)) else: print(render_text(payload) if render_text else payload) def _format_watch_text(rule: Dict[str, Any]) -> str: extra = [] if rule.get("days"): extra.append(f"최근 {rule['days']}일") if rule.get("exclude_words"): extra.append("제외=" + ", ".join(rule["exclude_words"])) if rule.get("template"): extra.append(f"template={rule['template']}") if rule.get("schedule", {}).get("label"): extra.append(f"schedule={rule['schedule']['label']}") lines = [f"- {rule['name']}: {rule['raw_query']}" + (f" ({'; '.join(extra)})" if extra else "")] if rule.get("label"): lines.append(f" label: {rule['label']}") if rule.get("tags"): lines.append(" tags: " + ", ".join(rule["tags"])) if rule.get("context"): lines.append(f" context: {rule['context']}") return "\n".join(lines) def _prompt_required(prompt: str, *, secret: bool = False) -> str: while True: value = (getpass.getpass(prompt) if secret else input(prompt)).strip() if value: return value print("값이 비어 있습니다. 다시 입력해 주세요.") def _validate_credential_value(name: str, value: str) -> str: cleaned = str(value or "").strip() if not cleaned: raise ValueError(f"{name} 값이 비어 있습니다.") if any(ch.isspace() for ch in cleaned): raise ValueError(f"{name} 값에 공백이 포함되어 있습니다. 복사 과정에서 줄바꿈/공백이 들어갔는지 확인해 주세요.") if len(cleaned) < 5: raise ValueError(f"{name} 값이 너무 짧습니다. 네이버 개발자센터에서 발급받은 값을 다시 확인해 주세요.") return cleaned def _perform_live_check(query: str, *, limit: int = 1) -> Dict[str, Any]: client_id, client_secret, timeout, _ = get_runtime_credentials() return fetch_news( client_id=client_id, client_secret=client_secret, search_query=query, exclude_words=[], limit=limit, days=None, timeout=timeout, ) def _render_setup_success(timeout: int, *, live_checked: bool = False, live_query: str | None = None) -> str: lines = [ "네이버 Search API 자격증명을 저장했습니다.", "- 저장 위치: data/config.json", f"- 요청 타임아웃: {timeout}초", ] if live_checked: lines.append(f"- 라이브 API 확인: 성공 ({live_query or DEFAULT_TEST_QUERY})") else: lines.append("- 라이브 API 확인: 생략") lines.extend( [ "- 다음 단계: python scripts/naver_news_briefing.py check-credentials --json", "- 그 다음: python scripts/naver_news_briefing.py search \"최근 3일 반도체 뉴스 브리핑\"", ] ) return "\n".join(lines) def cmd_setup(args: argparse.Namespace) -> int: client_id = (args.client_id or "").strip() client_secret = (args.client_secret or "").strip() if not client_id: print("client_id가 없어 대화형 입력으로 전환합니다.") client_id = _prompt_required("네이버 client_id: ") if not client_secret: print("client_secret이 없어 대화형 입력으로 전환합니다.") client_secret = _prompt_required("네이버 client_secret: ", secret=True) client_id = _validate_credential_value("client_id", client_id) client_secret = _validate_credential_value("client_secret", client_secret) set_credentials(client_id, client_secret, args.timeout) live_checked = False live_query = args.test_search or (DEFAULT_TEST_QUERY if args.live_check else None) if live_query: result = _perform_live_check(live_query, limit=1) live_checked = True if args.json: print(json.dumps({ "saved": True, "timeout": args.timeout, "live_check": { "query": live_query, "ok": True, "displayed": result.get("displayed", 0), "total": result.get("total", 0), }, }, ensure_ascii=False, indent=2)) return 0 print(_render_setup_success(args.timeout, live_checked=live_checked, live_query=live_query)) return 0 def _render_check_credentials_text(payload: Dict[str, Any]) -> str: lines = [ "네이버 Search API 자격증명이 설정되어 있습니다." if payload["configured"] else "아직 네이버 Search API 자격증명이 설정되지 않았습니다.", f"- client_id 저장 여부: {'예' if payload['client_id_present'] else '아니오'}", f"- client_secret 저장 여부: {'예' if payload['client_secret_present'] else '아니오'}", f"- 요청 타임아웃: {payload['timeout']}초", ] live_check = payload.get("live_check") if isinstance(live_check, dict): if live_check.get("ok"): lines.append(f"- 라이브 API 확인: 성공 ({live_check.get('query')})") lines.append(f"- 테스트 결과: total={live_check.get('total', 0)}, displayed={live_check.get('displayed', 0)}") else: lines.append(f"- 라이브 API 확인: 실패 ({live_check.get('query')})") if live_check.get("error"): lines.append(f"- 오류: {live_check['error']}") if payload["configured"]: lines.append("- 다음 단계: search / watch / brief-multi / plan-save 를 실행하면 됩니다.") else: lines.append("- 먼저 실행: python scripts/naver_news_briefing.py setup") lines.append("- 확인: python scripts/naver_news_briefing.py check-credentials --json") return "\n".join(lines) def cmd_check_credentials(args: argparse.Namespace) -> int: client_id, client_secret, timeout, _ = get_runtime_credentials() ok = bool(client_id and client_secret) payload: Dict[str, Any] = {"configured": ok, "client_id_present": bool(client_id), "client_secret_present": bool(client_secret), "timeout": timeout} if ok and args.live_check: query = args.query or DEFAULT_TEST_QUERY try: result = _perform_live_check(query, limit=1) payload["live_check"] = { "query": query, "ok": True, "displayed": result.get("displayed", 0), "total": result.get("total", 0), } except Exception as exc: payload["live_check"] = {"query": query, "ok": False, "error": str(exc)} ok = False print(json.dumps(payload, ensure_ascii=False, indent=2) if args.json else _render_check_credentials_text(payload)) return 0 if ok else 1 def _render_missing_credentials_guidance() -> str: return "\n".join( [ "네이버 Search API 자격증명이 아직 설정되지 않았습니다.", "먼저 최초 온보딩(setup)을 완료해 주세요.", "", "실행 순서:", "1) python scripts/naver_news_briefing.py setup --client-id YOUR_ID --client-secret YOUR_SECRET", "2) python scripts/naver_news_briefing.py check-credentials --json", "3) 그 다음 search / watch / brief-multi / plan-save 명령을 실행", "", "참고: 자격증명은 data/config.json 에 저장되며 Windows에서는 가능하면 DPAPI로 보호합니다.", ] ) def _dedupe_preserve_order(values: List[str]) -> List[str]: seen = set() ordered: List[str] = [] for value in values: if value in seen: continue seen.add(value) ordered.append(value) return ordered def _format_missing_named_resource(resource_label: str, missing_name: str, available_names: List[str]) -> str: lines = [f"등록된 {resource_label}을(를) 찾지 못했습니다: {missing_name}"] if available_names: lines.append("현재 등록된 항목: " + ", ".join(available_names)) else: lines.append("현재 등록된 항목이 없습니다.") return "\n".join(lines) def _format_exception_message(exc: Exception) -> str: message = str(exc).strip() or exc.__class__.__name__ if MISSING_CREDENTIALS_ERROR in message: return _render_missing_credentials_guidance() if "keyword group not found:" in message: missing_name = message.split("keyword group not found:", 1)[1].strip().strip("'") available = [group.get("name") for group in list_groups() if group.get("name")] return _format_missing_named_resource("키워드 그룹", missing_name, available) if "watch rule not found:" in message: missing_name = message.split("watch rule not found:", 1)[1].strip().strip("'") available = [rule.get("name") for rule in list_rules() if rule.get("name")] return _format_missing_named_resource("watch rule", missing_name, available) return f"ERROR: {message}" def run_search(query: str, *, limit: int, days: int | None, as_json: bool) -> int: intent = build_intent(query, limit=limit, days=days) client_id, client_secret, timeout, _ = get_runtime_credentials() result = fetch_news(client_id=client_id, client_secret=client_secret, search_query=intent.search_query, exclude_words=intent.exclude_words, limit=intent.limit, days=intent.days, timeout=timeout) result["intent"] = {"db_keyword": intent.db_keyword, "fetch_key": intent.fetch_key, "raw_query": intent.raw_query} if as_json: print(json.dumps(result, ensure_ascii=False, indent=2)) else: print("\n".join(_brief_lines(result))) return 0 def cmd_search(args: argparse.Namespace) -> int: return run_search(args.query, limit=args.limit, days=args.days, as_json=args.json) def cmd_watch_add(args: argparse.Namespace) -> int: intent = build_intent(args.query, limit=args.limit, days=args.days) rule = add_rule( name=args.name, raw_query=intent.raw_query, search_query=intent.search_query, db_keyword=intent.db_keyword, exclude_words=intent.exclude_words, fetch_key=intent.fetch_key, days=intent.days, limit=intent.limit, label=args.label, tags=args.tag, context=args.context, template=args.template, ) _print_payload(rule, as_json=args.json, render_text=_format_watch_text) return 0 def cmd_watch_list(args: argparse.Namespace) -> int: rules = list_rules() if args.json: print(json.dumps(rules, ensure_ascii=False, indent=2)) return 0 if not rules: print("등록된 watch rule이 없습니다.") return 0 print("\n\n".join(_format_watch_text(rule) for rule in rules)) return 0 def cmd_watch_remove(args: argparse.Namespace) -> int: deleted = remove_rule(args.name_or_id) if deleted: print(f"삭제 완료: {args.name_or_id}") return 0 print(f"삭제 대상 없음: {args.name_or_id}") return 1 def _run_rule(rule: Dict[str, Any]) -> Dict[str, Any]: client_id, client_secret, timeout, _ = get_runtime_credentials() result = fetch_news(client_id=client_id, client_secret=client_secret, search_query=rule["search_query"], exclude_words=rule["exclude_words"], limit=rule["limit"], days=rule.get("days"), timeout=timeout) new_items = mark_seen(rule["id"], result["items"]) return { "rule": rule, "summary": { "query": result["query"], "total": result["total"], "displayed": result["displayed"], "new_count": len(new_items), "filtered_out": result["filtered_out"], "too_old": result["too_old"], }, "new_items": new_items, "all_items": result["items"], } def _format_watch_status_lines(entry: Dict[str, Any]) -> List[str]: rule = entry["rule"] summary = entry["summary"] items = entry.get("all_items", []) now = datetime.now().strftime("%m월 %d일 %H:%M") query = summary.get("query") or rule.get("search_query") or rule.get("name") current_top_count = len(items) new_count = summary["new_count"] lines = [ f"## {rule['name']}", f"- {now} 기준, '{query}' 관련 상위 {current_top_count}건을 확인했고 이번 체크에서 신규 기사 {new_count}건이 추가됐습니다.", f"- 전체 검색 결과: {summary.get('total', 0)}건", f"- 이번 확인 신규 기사: {new_count}건", f"- 현재 검색 상위 기사 수: {current_top_count}건", ] if items: latest_pub = next((item.get("pub_date_iso") or item.get("pub_date") for item in items if item.get("pub_date_iso") or item.get("pub_date")), None) if latest_pub: lines.append(f"- 현재 기준 최신 기사 시각: {latest_pub}") return lines def cmd_watch_check(args: argparse.Namespace) -> int: targets = [get_rule(args.name_or_id)] if args.name_or_id else list_rules() payload = [_run_rule(rule) for rule in targets] if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 if not payload: print("체크할 watch rule이 없습니다.") return 0 if getattr(args, "announce_text", False): if len(payload) != 1: print("NO_REPLY") return 0 entry = payload[0] new_items = entry.get("new_items", []) if not new_items: print("NO_REPLY") return 0 summary = entry.get("summary", {}) new_count = summary.get("new_count", len(new_items)) latest_iso = new_items[0].get("pub_date_iso") latest_text = latest_iso.replace("T", " ")[:16] if latest_iso else "최신 없음" preview_items = new_items[:ANNOUNCE_PREVIEW_LIMIT] lines = [ f"확인 요약: 새 뉴스 {new_count}건, 현재 검색 상위 {summary.get('displayed', 0)}건 / 전체 {summary.get('total', 0)}건, 최신 {latest_text}" ] for item in preview_items: title = _strip_html(item.get("title", "제목 없음")) publisher = item.get("publisher") or "정보 없음" pub_iso = item.get("pub_date_iso") pub_text = pub_iso.replace("T", " ")[:16] if pub_iso else item.get("pub_date", "시간 없음") link = item.get("link") or item.get("original_link") or "" lines.append(f"- {title} / {publisher} / {pub_text} / {link}") extra_count = max(0, new_count - len(preview_items)) if extra_count > 0: lines.append(f"- 외 추가 새 뉴스 {extra_count}건") print("\n".join(lines)) return 0 lines: List[str] = [] for entry in payload: rule = entry["rule"] lines.extend(_format_watch_status_lines(entry)) lines.extend(_brief_lines({ "query": rule["search_query"], "exclude_words": rule["exclude_words"], "days": rule.get("days"), "total": entry["summary"]["total"], "displayed": len(entry["new_items"]), "filtered_out": entry["summary"]["filtered_out"], "too_old": entry["summary"]["too_old"], "items": entry["new_items"], }, title=f"watch: {rule['name']} 신규 기사 요약")) lines.append("") print("\n".join(lines).strip()) return 0 def _format_group_text(group: Dict[str, Any]) -> str: lines = [f"- {group['name']} ({group['query_count']}개 쿼리)"] if group.get("label"): lines.append(f" label: {group['label']}") if group.get("tags"): lines.append(" tags: " + ", ".join(group["tags"])) if group.get("template"): lines.append(f" template: {group['template']}") if group.get("schedule", {}).get("label"): lines.append(f" schedule: {group['schedule']['label']}") if group.get("context"): lines.append(f" context: {group['context']}") for idx, query in enumerate(group.get("queries", []), start=1): lines.append(f" {idx}. {query}") return "\n".join(lines) def cmd_group_add(args: argparse.Namespace) -> int: group = create_group(name=args.name, queries=args.query, label=args.label, tags=args.tag, context=args.context, template=args.template) _print_payload(group, as_json=args.json, render_text=_format_group_text) return 0 def cmd_group_list(args: argparse.Namespace) -> int: groups = [get_group(args.name_or_id)] if args.name_or_id else list_groups() if args.json: print(json.dumps(groups if not args.name_or_id else groups[0], ensure_ascii=False, indent=2)) return 0 if not groups: print("등록된 키워드 그룹이 없습니다.") return 0 print("\n\n".join(_format_group_text(group) for group in groups)) return 0 def cmd_group_remove(args: argparse.Namespace) -> int: deleted = remove_group(args.name_or_id) if deleted: print(f"삭제 완료: {args.name_or_id}") return 0 print(f"삭제 대상 없음: {args.name_or_id}") return 1 def cmd_group_update(args: argparse.Namespace) -> int: tags = None if args.tag is not None or args.clear_tags: tags = [] if args.clear_tags else args.tag group = update_group(args.name_or_id, label=args.label, context=args.context, tags=tags, template=args.template, replace_queries=args.set_query, add_queries=args.add_query, remove_queries=args.remove_query) _print_payload(group, as_json=args.json, render_text=_format_group_text) return 0 def _run_query_entry(query: str, *, limit: int, days: int | None, group: Dict[str, Any] | None = None, label: str | None = None, context: str | None = None) -> Dict[str, Any]: intent = build_intent(query, limit=limit, days=days) client_id, client_secret, timeout, _ = get_runtime_credentials() result = fetch_news(client_id=client_id, client_secret=client_secret, search_query=intent.search_query, exclude_words=intent.exclude_words, limit=intent.limit, days=intent.days, timeout=timeout) result["intent"] = {"db_keyword": intent.db_keyword, "fetch_key": intent.fetch_key, "raw_query": intent.raw_query} return { "query": query, "search_query": intent.search_query, "exclude_words": intent.exclude_words, "days": intent.days, "group_name": group["name"] if group else None, "label": label or (group.get("label") if group else None), "tags": group.get("tags", []) if group else [], "context": context or (group.get("context") if group else None), "result": result, } def cmd_brief_multi(args: argparse.Namespace) -> int: entries: List[Dict[str, Any]] = [] groups: List[Dict[str, Any]] = [] template = args.template for group_name in args.group or []: group = get_group(group_name) groups.append(group) if not args.template and group.get("template"): template = group["template"] for query in group["queries"]: entries.append(_run_query_entry(query, limit=args.limit, days=args.days, group=group)) for query in args.query or []: entries.append(_run_query_entry(query, limit=args.limit, days=args.days)) if not entries: raise ValueError("brief-multi에는 --query 또는 --group이 최소 1개 이상 필요합니다.") payload = build_combined_payload(entries, template=template, source_groups=groups) if args.json: print(render_combined_json(payload)) else: print(render_combined_text(payload)) return 0 def cmd_plan(args: argparse.Namespace) -> int: plan = parse_automation_request(args.request) _print_payload(plan_to_dict(plan), as_json=args.json, render_text=lambda _: render_plan_text(plan)) return 0 def cmd_integration_plan(args: argparse.Namespace) -> int: bundle = build_integration_bundle( args.request, skill_dir=args.skill_dir or str(Path(__file__).resolve().parents[1]), assistant_channel=args.channel, ) if args.output: output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w", encoding="utf-8") as fp: json.dump(bundle, fp, ensure_ascii=False, indent=2) _print_payload(bundle, as_json=args.json, render_text=render_integration_bundle_text) return 0 def cmd_plan_save(args: argparse.Namespace) -> int: plan = parse_automation_request(args.request) created: Dict[str, Any] = {"plan": plan_to_dict(plan), "created": []} name = args.name or plan.name_hint label = args.label or ("아침 브리핑" if plan.template == "morning-briefing" else None) tags = list(args.tag or []) if plan.schedule.kind != "manual": tags.append(plan.schedule.kind) if plan.watch_intent != "none": tags.append("watch") if plan.query_mode == "group": tags.append("group") tags = _dedupe_preserve_order(tags) if not plan.queries: raise ValueError("저장 가능한 주제 키워드를 찾지 못했습니다. 요청에 관심 주제를 포함해 주세요.") if args.as_type == "watch" and plan.query_mode == "group": raise ValueError("여러 주제를 포함한 그룹형 브리핑 요청입니다. --as group 으로 저장해 주세요.") if args.as_type == "group" or plan.query_mode == "group": group = create_group( name=name, queries=plan.queries, label=label, tags=tags, context=plan.raw_request, template=plan.template, schedule=plan_to_dict(plan)["schedule"], operator_hints=plan_to_dict(plan)["operator_hints"], ) created["created"].append({"type": "group", "value": group}) else: intent = build_intent(plan.primary_query or plan.queries[0]) rule = add_rule( name=name, raw_query=intent.raw_query, search_query=intent.search_query, db_keyword=intent.db_keyword, exclude_words=intent.exclude_words, fetch_key=intent.fetch_key, days=intent.days, limit=intent.limit, label=label, tags=tags, context=plan.raw_request, template=plan.template, schedule=plan_to_dict(plan)["schedule"], operator_hints=plan_to_dict(plan)["operator_hints"], ) created["created"].append({"type": "watch", "value": rule}) _print_payload(created, as_json=args.json, render_text=lambda payload: render_plan_text(plan) + "\n- 저장 결과:\n" + "\n".join(f" - {item['type']}: {item['value']['name']}" for item in payload['created'])) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Naver news briefing skill CLI") sub = parser.add_subparsers(dest="command", required=True) p = sub.add_parser("setup", help="네이버 API 자격증명 저장") p.add_argument("--client-id") p.add_argument("--client-secret") p.add_argument("--timeout", type=int, default=15) p.add_argument("--live-check", action="store_true", help="저장 직후 실제 API 호출로 자격증명을 확인") p.add_argument("--test-search", help="저장 직후 테스트할 검색어 (설정 시 라이브 체크 수행)") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_setup) p = sub.add_parser("check-credentials", help="자격증명 상태 확인") p.add_argument("--json", action="store_true") p.add_argument("--live-check", action="store_true", help="실제 API 호출까지 포함해 확인") p.add_argument("--query", help="라이브 체크용 검색어") p.set_defaults(func=cmd_check_credentials) p = sub.add_parser("search", help="원샷 뉴스 검색/브리핑") p.add_argument("query") p.add_argument("--limit", type=int, default=10) p.add_argument("--days", type=int) p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_search) p = sub.add_parser("watch-add", help="watch rule 추가") p.add_argument("name") p.add_argument("query") p.add_argument("--limit", type=int, default=10) p.add_argument("--days", type=int) p.add_argument("--label") p.add_argument("--tag", action="append") p.add_argument("--context") p.add_argument("--template", choices=supported_templates()) p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_watch_add) p = sub.add_parser("watch-list", help="watch rule 목록") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_watch_list) p = sub.add_parser("watch-remove", help="watch rule 삭제") p.add_argument("name_or_id") p.set_defaults(func=cmd_watch_remove) p = sub.add_parser("watch-check", help="watch rule 신규 기사 체크") p.add_argument("name_or_id", nargs="?") p.add_argument("--json", action="store_true") p.add_argument("--announce-text", action="store_true", help="cron/announce friendly plain text output") p.set_defaults(func=cmd_watch_check) p = sub.add_parser("group-add", help="키워드 그룹 추가") p.add_argument("name") p.add_argument("query", nargs="+") p.add_argument("--label") p.add_argument("--tag", action="append") p.add_argument("--context") p.add_argument("--template", choices=supported_templates()) p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_group_add) p = sub.add_parser("group-list", help="키워드 그룹 조회/목록") p.add_argument("name_or_id", nargs="?") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_group_list) p = sub.add_parser("group-remove", help="키워드 그룹 삭제") p.add_argument("name_or_id") p.set_defaults(func=cmd_group_remove) p = sub.add_parser("group-update", help="키워드 그룹 수정") p.add_argument("name_or_id") p.add_argument("--label") p.add_argument("--context") p.add_argument("--template", choices=supported_templates()) p.add_argument("--tag", action="append") p.add_argument("--clear-tags", action="store_true") p.add_argument("--set-query", action="append") p.add_argument("--add-query", action="append") p.add_argument("--remove-query", action="append") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_group_update) p = sub.add_parser("brief-multi", help="여러 쿼리/그룹을 묶어 한 번에 브리핑") p.add_argument("--query", action="append") p.add_argument("--group", action="append") p.add_argument("--limit", type=int, default=5) p.add_argument("--days", type=int) p.add_argument("--template", choices=supported_templates()) p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_brief_multi) p = sub.add_parser("plan", help="채팅형 자연어 요청을 자동화 계획으로 변환") p.add_argument("request") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_plan) p = sub.add_parser("integration-plan", help="자연어 요청을 OpenClaw/cron 연동 번들로 변환") p.add_argument("request") p.add_argument("--channel", default="telegram") p.add_argument("--skill-dir") p.add_argument("--output") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_integration_plan) p = sub.add_parser("plan-save", help="자연어 요청을 해석해 watch/group 설정으로 저장") p.add_argument("request") p.add_argument("--name") p.add_argument("--as", dest="as_type", choices=["watch", "group"]) p.add_argument("--label") p.add_argument("--tag", action="append") p.add_argument("--json", action="store_true") p.set_defaults(func=cmd_plan_save) return parser def main(argv: List[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) try: return args.func(args) except Exception as exc: print(_format_exception_message(exc), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/query_utils.py from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta from typing import List, Tuple # Adapted from upstream core.query_parser.py _REQUEST_PHRASES = [ "뉴스", "브리핑", "브리핑해줘", "브리핑 해줘", "요약", "요약해줘", "요약해 줘", "검색", "검색해줘", "검색해 줘", "찾아줘", "찾아 줘", "알려줘", "알려 줘", "정리해줘", "정리해 줘", "보여줘", "보여 줘", "뽑아줘", "뽑아 줘", "체크해줘", "체크해 줘", "모아줘", "모아 줘", "보고싶어", "보고 싶어", "궁금해", "궁금한", "관련해서", "관련 기사", "관련 뉴스", "관련", "대해서", "대해", "관한", "중심으로", "위주로", "위주", "핵심만", "핵심", "동향", ] _RECENT_TOKEN_DEFAULTS = { "오늘": 1, "금일": 1, "어제": 2, "최근": 7, "최신": 7, "요즘": 7, "이번주": 7, "이번 주": 7, "지난주": 14, "지난 주": 14, "일주일": 7, "한주": 7, "한 주": 7, "보름": 15, "한달": 30, "한 달": 30, "한달간": 30, "한 달간": 30, } _TOKEN_STOPWORDS = { "관련", "관련해서", "관련된", "대해", "대해서", "대한", "기사", "뉴스", "브리핑", "검색", "요약", "정리", "핵심", "동향", "위주", "중심", "중심으로", "위주로", "알려줘", "찾아줘", "보여줘", "정리해줘", "요약해줘", "브리핑해줘", "검색해줘", "해줘", "만", "최근", "최신", "오늘", "금일", "어제", "이번주", "지난주", "일주일", "한주", "한달", "한달간", "보름", } _KOREAN_PARTICLES = [ "에서는", "에선", "으로는", "로는", "에게는", "한테는", "과는", "와는", "으로", "에서", "에게", "한테", "까지", "부터", "처럼", "보다", "마저", "조차", "이라도", "라도", "이나", "나", "은", "는", "이", "가", "을", "를", "에", "의", "과", "와", "도", "만", ] def _split_query_tokens(raw: str) -> Tuple[List[str], List[str]]: parts = str(raw or "").split() if not parts: return [], [] positive_words: List[str] = [] exclude_words: List[str] = [] for token in parts: if token.startswith("-"): if len(token) > 1: exclude_words.append(token[1:]) continue positive_words.append(token) return positive_words, exclude_words def parse_tab_query(raw: str) -> Tuple[str, List[str]]: positive_words, exclude_words = _split_query_tokens(raw) db_keyword = positive_words[0] if positive_words else "" return db_keyword, exclude_words def parse_search_query(raw: str) -> Tuple[str, List[str]]: positive_words, exclude_words = _split_query_tokens(raw) search_query = " ".join(positive_words) return search_query, exclude_words def build_fetch_key(search_keyword: str, exclude_words: List[str]) -> str: normalized_keyword = (search_keyword or "").strip().lower() normalized_excludes = sorted( { word.strip().lower() for word in (exclude_words or []) if isinstance(word, str) and word.strip() } ) return f"{normalized_keyword}|{'|'.join(normalized_excludes)}" @dataclass(frozen=True) class QueryIntent: raw_query: str search_query: str db_keyword: str exclude_words: List[str] fetch_key: str days: int | None limit: int def detect_recent_days(raw: str) -> int | None: lowered = str(raw or "").lower() import re patterns = [ r"최근\s*(\d+)\s*일", r"(\d+)\s*일\s*(내|이내)", r"최근\s*(\d+)\s*주", r"(\d+)\s*주\s*(내|이내)", r"최근\s*(\d+)\s*(?:개월|달)", r"(\d+)\s*(?:개월|달)\s*(내|이내)", ] for pattern in patterns: m = re.search(pattern, raw) if not m: continue value = int(m.group(1)) if "주" in pattern: value *= 7 elif "개월" in pattern or "달" in pattern: value *= 30 return max(1, min(365, value)) for token, days in _RECENT_TOKEN_DEFAULTS.items(): if token in raw: return days if "today" in lowered: return 1 if "this week" in lowered or "recent" in lowered or "latest" in lowered: return 7 if "last week" in lowered: return 14 return None def _normalize_spacing(raw: str) -> str: import re text = str(raw or "").strip() text = re.sub(r"[\[\]{}<>\"'“”‘’,:;!?/~`]+", " ", text) text = text.replace("·", " ").replace("…", " ") text = re.sub(r"\s+", " ", text) return text.strip() def _apply_exclude_phrases(raw: str) -> str: import re text = _normalize_spacing(raw) patterns = [ r"([가-힣A-Za-z0-9+#./_-]+(?:\s+[가-힣A-Za-z0-9+#./_-]+){0,2})\s*(?:말고|빼고|제외|제외하고|제외한|없이)", r"([가-힣A-Za-z0-9+#./_-]+(?:\s+[가-힣A-Za-z0-9+#./_-]+){0,2})\s*(?:은|는|을|를)?\s*빼면", ] def repl(match: re.Match[str]) -> str: words = [w for w in match.group(1).split() if w] if not words: return " " return " " + " ".join(f"-{word}" for word in words) + " " for pattern in patterns: text = re.sub(pattern, repl, text) return _normalize_spacing(text) def _strip_particle(token: str) -> str: stripped = token.strip() if stripped.startswith("-"): head = "-" stripped = stripped[1:] else: head = "" if not stripped: return token for particle in _KOREAN_PARTICLES: if stripped.endswith(particle) and len(stripped) > len(particle) + 1: stripped = stripped[: -len(particle)] break return head + stripped def _normalize_token(token: str) -> str: import re token = _strip_particle(token) token = token.strip("- ") if token == "-" else token.strip() token = re.sub(r"^[^\w가-힣+#]+|[^\w가-힣+#-]+$", "", token) token = token.strip() if token.startswith("-"): body = token[1:].strip() body = re.sub(r"^[^\w가-힣+#]+|[^\w가-힣+#-]+$", "", body) return f"-{body}" if body else "" return token def clean_natural_query(raw: str) -> str: import re stripped = _apply_exclude_phrases(raw) for token in _REQUEST_PHRASES: stripped = stripped.replace(token, " ") stripped = re.sub(r"\b(today|latest|recent|this week|last week)\b", " ", stripped, flags=re.IGNORECASE) stripped = re.sub(r"최근\s*\d+\s*(?:일|주|개월|달)", " ", stripped) stripped = re.sub(r"\d+\s*(?:일|주|개월|달)\s*(?:내|이내)", " ", stripped) stripped = re.sub(r"(?:일주일|한\s*주|한\s*달|한달|보름|이번\s*주|지난\s*주)", " ", stripped) stripped = re.sub(r"\b\d+\b", " ", stripped) stripped = _normalize_spacing(stripped) normalized_tokens: List[str] = [] for raw_token in stripped.split(): negative = raw_token.startswith("-") body = raw_token[1:] if negative else raw_token body = re.sub(r"^[^\w가-힣+#]+|[^\w가-힣+#]+$", "", body) body = _strip_particle(body).strip() body = body.strip("- ") if not body or body in _TOKEN_STOPWORDS: continue token = f"-{body}" if negative else body normalized_tokens.append(token) deduped: List[str] = [] seen = set() for token in normalized_tokens: key = token.lower() if key in seen: continue seen.add(key) deduped.append(token) return " ".join(deduped) def build_intent(raw_query: str, *, limit: int = 10, days: int | None = None) -> QueryIntent: cleaned = clean_natural_query(raw_query) detected_days = days if days is not None else detect_recent_days(raw_query) search_query, exclude_words = parse_search_query(cleaned) db_keyword, _ = parse_tab_query(cleaned) if not search_query: raise ValueError("최소 1개 이상의 일반 키워드가 필요합니다.") return QueryIntent( raw_query=raw_query, search_query=search_query, db_keyword=db_keyword or search_query, exclude_words=exclude_words, fetch_key=build_fetch_key(search_query, exclude_words), days=detected_days, limit=max(1, min(100, int(limit))), ) def cutoff_iso(days: int | None, now: datetime | None = None) -> str | None: if not days: return None base = now or datetime.now() return (base - timedelta(days=days)).isoformat(timespec="seconds") FILE:scripts/tests/conftest.py from __future__ import annotations import sys from pathlib import Path SCRIPTS_DIR = Path(__file__).resolve().parents[1] if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) FILE:scripts/tests/test_automation_plans.py import json import naver_news_briefing as cli from automation_plans import build_integration_bundle, parse_automation_request class DummyArgs: def __init__(self, **kwargs): self.__dict__.update(kwargs) def test_parse_monitoring_interval_plan(): plan = parse_automation_request("반도체 뉴스 1시간마다 모니터링해줘") assert plan.action == "monitor" assert plan.schedule.kind == "interval" assert plan.schedule.interval_minutes == 60 assert plan.primary_query == "반도체" assert plan.template == "watch-alert" assert plan.operator_hints.storage_target == "watch" assert any("watch-add" in cmd for cmd in plan.suggested_commands) def test_parse_daily_briefing_group_plan(): plan = parse_automation_request("매일 아침 7시에 반도체, AI 데이터센터 뉴스 브리핑해줘") assert plan.action == "briefing" assert plan.query_mode == "group" assert plan.schedule.kind == "daily" assert plan.schedule.time == "07:00" assert len(plan.queries) == 2 assert plan.template == "morning-briefing" assert any("group-add" in cmd for cmd in plan.suggested_commands) def test_parse_exclude_and_continuous_watch_plan(): plan = parse_automation_request("증권사 리포트 빼고 삼성전자 뉴스 계속 체크해줘") assert plan.action == "monitor" assert plan.primary_query == "삼성전자 -증권사 -리포트" assert plan.schedule.kind == "interval" assert plan.schedule.interval_minutes == 15 assert plan.watch_intent == "continuous" def test_cmd_plan_json(capsys): args = DummyArgs(request="반도체 뉴스 1시간마다 모니터링해줘", json=True) assert cli.cmd_plan(args) == 0 payload = json.loads(capsys.readouterr().out) assert payload["schedule"]["interval_minutes"] == 60 assert payload["primary_query"] == "반도체" assert payload["operator_hints"]["storage_target"] == "watch" def test_cmd_plan_save_creates_watch(monkeypatch, capsys): monkeypatch.setattr(cli, "add_rule", lambda **kwargs: {"name": kwargs["name"], "search_query": kwargs["search_query"], "template": kwargs["template"], "tags": kwargs["tags"], "schedule": kwargs["schedule"]}) args = DummyArgs(request="반도체 뉴스 1시간마다 모니터링해줘", name="semi-hourly", as_type="watch", label=None, tag=None, json=True) assert cli.cmd_plan_save(args) == 0 payload = json.loads(capsys.readouterr().out) value = payload["created"][0]["value"] assert payload["created"][0]["type"] == "watch" assert value["name"] == "semi-hourly" assert value["template"] == "watch-alert" assert "interval" in value["tags"] assert value["schedule"]["interval_minutes"] == 60 def test_build_integration_bundle_for_watch(tmp_path): skill_dir = tmp_path / "skills" / "naver-news-briefing" skill_dir.mkdir(parents=True) bundle = build_integration_bundle( "반도체 뉴스 1시간마다 모니터링해줘", skill_dir=skill_dir, assistant_channel="telegram", ) assert bundle["storage"]["target"] == "watch" assert bundle["storage"]["name"] == "반도체-watch" assert "plan-save" in bundle["storage"]["save_command"] assert "watch-check 반도체-watch --json" in bundle["runner"]["command"] assert bundle["automation"]["schedule"]["cron"] == "0 */1 * * *" assert bundle["automation"]["cron_line"].startswith("0 */1 * * * cd ") assert "telegram 채널" in bundle["automation"]["system_event_text"] def test_cmd_integration_plan_json(tmp_path, capsys): skill_dir = tmp_path / "skills" / "naver-news-briefing" skill_dir.mkdir(parents=True) args = DummyArgs( request="매일 아침 7시에 반도체랑 AI 데이터센터 뉴스 브리핑해줘", channel="telegram", skill_dir=str(skill_dir), output=None, json=True, ) assert cli.cmd_integration_plan(args) == 0 payload = json.loads(capsys.readouterr().out) assert payload["storage"]["target"] == "group" assert payload["plan"]["schedule"]["time"] == "07:00" assert "brief-multi --group" in payload["runner"]["command"] assert payload["automation"]["schedule"]["cron"] == "0 7 * * *" def test_cmd_integration_plan_creates_output_parent_dirs(tmp_path, capsys): skill_dir = tmp_path / "skills" / "naver-news-briefing" skill_dir.mkdir(parents=True) output_path = tmp_path / "nested" / "plans" / "bundle.json" args = DummyArgs( request="반도체 뉴스 1시간마다 모니터링해줘", channel="telegram", skill_dir=str(skill_dir), output=str(output_path), json=True, ) assert cli.cmd_integration_plan(args) == 0 payload = json.loads(capsys.readouterr().out) assert output_path.exists() assert json.loads(output_path.read_text(encoding="utf-8"))["storage"]["target"] == payload["storage"]["target"] def test_cmd_plan_save_creates_group(monkeypatch, capsys): monkeypatch.setattr(cli, "create_group", lambda **kwargs: {"name": kwargs["name"], "queries": kwargs["queries"], "template": kwargs["template"], "schedule": kwargs["schedule"], "operator_hints": kwargs["operator_hints"]}) args = DummyArgs(request="반도체, AI 데이터센터 뉴스 매일 아침 7시에 브리핑해줘", name="morning-tech", as_type="group", label="아침 브리핑", tag=["테크"], json=True) assert cli.cmd_plan_save(args) == 0 payload = json.loads(capsys.readouterr().out) value = payload["created"][0]["value"] assert payload["created"][0]["type"] == "group" assert value["name"] == "morning-tech" assert len(value["queries"]) == 2 assert value["template"] == "morning-briefing" assert value["schedule"]["time"] == "07:00" assert value["operator_hints"]["storage_target"] == "group" def test_cmd_plan_save_dedupes_tags(monkeypatch, capsys): monkeypatch.setattr(cli, "add_rule", lambda **kwargs: {"name": kwargs["name"], "tags": kwargs["tags"]}) args = DummyArgs( request="반도체 뉴스 1시간마다 모니터링해줘", name="semi-hourly", as_type="watch", label=None, tag=["watch", "interval", "watch", "사용자태그"], json=True, ) assert cli.cmd_plan_save(args) == 0 payload = json.loads(capsys.readouterr().out) assert payload["created"][0]["value"]["tags"] == ["watch", "interval", "사용자태그"] def test_cmd_plan_save_rejects_group_plan_as_watch(): args = DummyArgs( request="반도체, AI 데이터센터 뉴스 매일 아침 7시에 브리핑해줘", name="morning-tech", as_type="watch", label=None, tag=None, json=True, ) try: cli.cmd_plan_save(args) except ValueError as exc: assert "--as group" in str(exc) else: raise AssertionError("ValueError expected") def test_main_reports_missing_group_with_available_names(monkeypatch, capsys): monkeypatch.setattr(cli, "get_group", lambda name: (_ for _ in ()).throw(KeyError(f"keyword group not found: {name}"))) monkeypatch.setattr(cli, "list_groups", lambda: [{"name": "market-watch"}, {"name": "morning-tech"}]) exit_code = cli.main(["group-list", "missing-group"]) captured = capsys.readouterr() assert exit_code == 1 assert "등록된 키워드 그룹을(를) 찾지 못했습니다: missing-group" in captured.err assert "market-watch, morning-tech" in captured.err def test_main_reports_missing_watch_rule_with_hint(monkeypatch, capsys): monkeypatch.setattr(cli, "get_rule", lambda name: (_ for _ in ()).throw(KeyError(f"watch rule not found: {name}"))) monkeypatch.setattr(cli, "list_rules", lambda: [{"name": "semi-hourly"}]) exit_code = cli.main(["watch-check", "unknown-watch"]) captured = capsys.readouterr() assert exit_code == 1 assert "등록된 watch rule을(를) 찾지 못했습니다: unknown-watch" in captured.err assert "semi-hourly" in captured.err FILE:scripts/tests/test_brief_multi.py import json import naver_news_briefing as cli from briefing_templates import build_combined_payload, render_combined_text class DummyArgs: def __init__(self, **kwargs): self.__dict__.update(kwargs) def test_build_combined_payload_and_render_text(): payload = build_combined_payload( [ { "query": "반도체 뉴스", "group_name": "market-watch", "label": "시장", "context": "아침 브리핑", "result": { "displayed": 1, "total": 10, "filtered_out": 0, "too_old": 0, "items": [{"title": "반도체 상승", "publisher": "연합", "link": "https://example.com/a"}], }, } ], template="analyst", source_groups=[{"name": "market-watch"}], ) text = render_combined_text(payload) assert "네이버 뉴스 멀티 브리핑 · analyst" in text assert "market-watch" in text assert "반도체 상승" in text def test_cmd_brief_multi_supports_group_and_query_json(monkeypatch, capsys): monkeypatch.setattr(cli, "get_group", lambda name: { "name": name, "label": "관심주제", "tags": ["태그"], "context": "체크 포인트", "queries": ["반도체 -광고"], }) def fake_run_query_entry(query, *, limit, days, group=None, label=None, context=None): return { "query": query, "group_name": group["name"] if group else None, "label": label or (group.get("label") if group else None), "context": context or (group.get("context") if group else None), "result": { "displayed": 1, "total": 3, "filtered_out": 0, "too_old": 0, "items": [{"title": f"{query} 기사", "publisher": "연합", "link": "https://example.com/x"}], }, } monkeypatch.setattr(cli, "_run_query_entry", fake_run_query_entry) args = DummyArgs(group=["market-watch"], query=["환율 뉴스"], limit=5, days=None, template="morning-briefing", json=True) assert cli.cmd_brief_multi(args) == 0 payload = json.loads(capsys.readouterr().out) assert payload["template"] == "morning-briefing" assert payload["entry_count"] == 2 assert payload["groups"][0]["name"] == "market-watch" def test_main_brief_multi_missing_group_is_operator_friendly(monkeypatch, capsys): monkeypatch.setattr(cli, "get_group", lambda name: (_ for _ in ()).throw(KeyError(f"keyword group not found: {name}"))) monkeypatch.setattr(cli, "list_groups", lambda: [{"name": "market-watch"}]) exit_code = cli.main(["brief-multi", "--group", "missing-group"]) captured = capsys.readouterr() assert exit_code == 1 assert "등록된 키워드 그룹을(를) 찾지 못했습니다: missing-group" in captured.err assert "market-watch" in captured.err FILE:scripts/tests/test_group_store.py import group_store from group_store import create_group, get_group, list_groups, remove_group, update_group def test_group_store_create_update_remove(tmp_path, monkeypatch): monkeypatch.setattr(group_store, "DB_PATH", tmp_path / "watch.db") group = create_group( name="market-watch", queries=["반도체 -광고", "AI 데이터센터 -주가"], label="시장 모니터링", tags=["시장", "테크", "시장"], context="아침 브리핑용", template="morning-briefing", schedule={"kind": "daily", "label": "매일 07:00", "time": "07:00"}, operator_hints={"storage_target": "group"}, ) assert group["query_count"] == 2 assert group["tags"] == ["시장", "테크"] assert group["template"] == "morning-briefing" assert group["schedule"]["time"] == "07:00" assert len(list_groups()) == 1 updated = update_group( "market-watch", add_queries=["배터리 공급망 -광고"], remove_queries=["AI 데이터센터 -주가"], tags=["시장", "모니터링"], template="analyst", ) assert updated["queries"] == ["반도체 -광고", "배터리 공급망 -광고"] assert updated["tags"] == ["시장", "모니터링"] assert updated["template"] == "analyst" assert get_group("market-watch")["query_count"] == 2 assert remove_group("market-watch") == 1 FILE:scripts/tests/test_naver_api.py from naver_api import clean_item, fetch_news class DummyResponse: def __init__(self, status_code, payload): self.status_code = status_code self._payload = payload self.text = "" def json(self): return self._payload class DummySession: def __init__(self, payload): self.payload = payload def get(self, url, *, headers, params, timeout): return DummyResponse(200, self.payload) SAMPLE = { "total": 3, "items": [ { "title": "반도체 시장 확대", "description": "좋은 뉴스", "link": "https://news.naver.com/a", "originallink": "https://example.com/a", "pubDate": "Wed, 25 Mar 2026 10:00:00 +0900", }, { "title": "광고성 기사", "description": "광고 포함", "link": "https://example.com/b", "originallink": "https://example.com/b", "pubDate": "Wed, 25 Mar 2026 10:00:00 +0900", }, ], } def test_clean_item_prefers_naver_link_and_strips_html(): item = clean_item(SAMPLE["items"][0]) assert item.title == "반도체 시장 확대" assert item.link == "https://news.naver.com/a" assert item.publisher == "example.com" def test_fetch_news_applies_exclude_words_and_limit(): result = fetch_news( client_id="id", client_secret="secret", search_query="반도체", exclude_words=["광고"], limit=5, timeout=5, session=DummySession(SAMPLE), ) assert result["displayed"] == 1 assert result["filtered_out"] == 1 assert result["items"][0]["title"] == "반도체 시장 확대" FILE:scripts/tests/test_query_utils.py from query_utils import build_intent, clean_natural_query, parse_search_query, parse_tab_query def test_parse_queries_match_upstream_policy(): assert parse_search_query("인공지능 AI -광고 -코인") == ("인공지능 AI", ["광고", "코인"]) assert parse_tab_query("인공지능 AI -광고 -코인") == ("인공지능", ["광고", "코인"]) def test_build_intent_detects_recent_days_and_strips_briefing_words(): intent = build_intent("최근 3일 반도체 뉴스 브리핑 -광고", limit=7) assert intent.search_query == "반도체" assert intent.exclude_words == ["광고"] assert intent.days == 3 assert intent.limit == 7 def test_clean_natural_query_handles_sentence_style_korean(): cleaned = clean_natural_query("삼성전자 관련해서 최근 일주일 핵심 뉴스만 브리핑해줘") assert cleaned == "삼성전자" def test_build_intent_converts_korean_exclude_phrases(): intent = build_intent("삼성전자 관련해서 증권사 리포트 말고 최근 일주일 핵심만 알려줘") assert intent.search_query == "삼성전자" assert intent.exclude_words == ["증권사", "리포트"] assert intent.days == 7 def test_build_intent_strips_particles_and_dedupes_tokens(): intent = build_intent("반도체를 반도체 관련 뉴스 찾아줘") assert intent.search_query == "반도체" assert intent.exclude_words == [] FILE:scripts/tests/test_watch_check_output.py import naver_news_briefing as cli class DummyArgs: def __init__(self, **kwargs): self.__dict__.update(kwargs) def test_cmd_watch_check_renders_friendlier_status(monkeypatch, capsys): monkeypatch.setattr(cli, "get_rule", lambda name: { "id": 1, "name": name, "search_query": "방미심위", "exclude_words": [], "days": None, }) monkeypatch.setattr(cli, "_run_rule", lambda rule: { "rule": rule, "summary": { "query": "방미심위", "total": 1068, "displayed": 10, "new_count": 2, "filtered_out": 0, "too_old": 0, }, "new_items": [ { "title": "새 기사 A", "publisher": "연합", "link": "https://example.com/a", "pub_date": "Fri, 27 Mar 2026 17:05:00 +0900", }, { "title": "새 기사 B", "publisher": "뉴스1", "link": "https://example.com/b", "pub_date": "Fri, 27 Mar 2026 16:50:00 +0900", }, ], "all_items": [ { "title": "새 기사 A", "publisher": "연합", "link": "https://example.com/a", "pub_date_iso": "2026-03-27T17:05:00+09:00", } ], }) args = DummyArgs(name_or_id="bangmisimwi", json=False) assert cli.cmd_watch_check(args) == 0 out = capsys.readouterr().out assert "관련 상위 1건을 확인했고 이번 체크에서 신규 기사 2건이 추가됐습니다." in out assert "전체 검색 결과: 1068건" in out assert "이번 확인 신규 기사: 2건" in out assert "현재 검색 상위 기사 수: 1건" in out assert "현재 기준 최신 기사 시각: 2026-03-27T17:05:00+09:00" in out assert "watch: bangmisimwi 신규 기사 요약" in out FILE:scripts/tests/test_watch_store.py import watch_store from watch_store import add_rule, get_rule, list_rules, mark_seen, remove_rule def test_watch_store_add_list_seen_remove(tmp_path, monkeypatch): monkeypatch.setattr(watch_store, "DB_PATH", tmp_path / "watch.db") rule = add_rule( name="semiconductor", raw_query="반도체 -광고", search_query="반도체", db_keyword="반도체", exclude_words=["광고"], fetch_key="반도체|광고", days=7, limit=10, label="반도체 감시", tags=["watch", "interval"], context="1시간마다 모니터링", template="watch-alert", schedule={"kind": "interval", "interval_minutes": 60, "label": "1시간마다"}, operator_hints={"storage_target": "watch", "delivery_format": "watch-json"}, ) assert rule["name"] == "semiconductor" assert rule["template"] == "watch-alert" assert rule["schedule"]["interval_minutes"] == 60 assert get_rule("semiconductor")["label"] == "반도체 감시" assert len(list_rules()) == 1 first = mark_seen(rule["id"], [{"link": "https://example.com/a", "pub_date_iso": "2026-03-25T10:00:00+09:00"}]) second = mark_seen(rule["id"], [{"link": "https://example.com/a", "pub_date_iso": "2026-03-25T10:00:00+09:00"}]) assert len(first) == 1 assert len(second) == 0 assert remove_rule("semiconductor") == 1 FILE:scripts/watch_store.py from __future__ import annotations import json import sqlite3 from contextlib import contextmanager from datetime import datetime from typing import Any, Dict, Iterable, List from _paths import DB_PATH, ensure_data_dir SCHEMA = """ CREATE TABLE IF NOT EXISTS watch_rules ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, raw_query TEXT NOT NULL, search_query TEXT NOT NULL, db_keyword TEXT NOT NULL, exclude_json TEXT NOT NULL, fetch_key TEXT NOT NULL, days INTEGER, limit_count INTEGER NOT NULL, label TEXT, tags_json TEXT NOT NULL DEFAULT '[]', context TEXT, template TEXT, schedule_json TEXT NOT NULL DEFAULT '{}', operator_hint_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_checked_at TEXT, last_new_count INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS seen_items ( watch_id INTEGER NOT NULL, link TEXT NOT NULL, published_at TEXT, first_seen_at TEXT NOT NULL, PRIMARY KEY (watch_id, link), FOREIGN KEY (watch_id) REFERENCES watch_rules(id) ON DELETE CASCADE ); """ @contextmanager def connect(): ensure_data_dir() conn = sqlite3.connect(DB_PATH) try: conn.execute("PRAGMA foreign_keys = ON") conn.executescript(SCHEMA) columns = {row[1] for row in conn.execute("PRAGMA table_info(watch_rules)").fetchall()} migrations = { "label": "ALTER TABLE watch_rules ADD COLUMN label TEXT", "tags_json": "ALTER TABLE watch_rules ADD COLUMN tags_json TEXT NOT NULL DEFAULT '[]'", "context": "ALTER TABLE watch_rules ADD COLUMN context TEXT", "template": "ALTER TABLE watch_rules ADD COLUMN template TEXT", "schedule_json": "ALTER TABLE watch_rules ADD COLUMN schedule_json TEXT NOT NULL DEFAULT '{}'", "operator_hint_json": "ALTER TABLE watch_rules ADD COLUMN operator_hint_json TEXT NOT NULL DEFAULT '{}'", } for column, sql in migrations.items(): if column not in columns: conn.execute(sql) yield conn conn.commit() finally: conn.close() def _normalize_tags(tags: Iterable[str] | None) -> List[str]: seen = set() result: List[str] = [] for tag in tags or []: value = str(tag or "").strip() if not value or value in seen: continue seen.add(value) result.append(value) return result def list_rules() -> List[Dict[str, Any]]: with connect() as conn: rows = conn.execute( "SELECT id, name, raw_query, search_query, db_keyword, exclude_json, fetch_key, days, limit_count, label, tags_json, context, template, schedule_json, operator_hint_json, created_at, updated_at, last_checked_at, last_new_count FROM watch_rules ORDER BY name" ).fetchall() result = [] for row in rows: result.append( { "id": row[0], "name": row[1], "raw_query": row[2], "search_query": row[3], "db_keyword": row[4], "exclude_words": json.loads(row[5]), "fetch_key": row[6], "days": row[7], "limit": row[8], "label": row[9], "tags": json.loads(row[10] or "[]"), "context": row[11], "template": row[12], "schedule": json.loads(row[13] or "{}"), "operator_hints": json.loads(row[14] or "{}"), "created_at": row[15], "updated_at": row[16], "last_checked_at": row[17], "last_new_count": row[18], } ) return result def add_rule(*, name: str, raw_query: str, search_query: str, db_keyword: str, exclude_words: List[str], fetch_key: str, days: int | None, limit: int, label: str | None = None, tags: Iterable[str] | None = None, context: str | None = None, template: str | None = None, schedule: Dict[str, Any] | None = None, operator_hints: Dict[str, Any] | None = None) -> Dict[str, Any]: now = datetime.now().isoformat(timespec="seconds") with connect() as conn: conn.execute( """ INSERT INTO watch_rules(name, raw_query, search_query, db_keyword, exclude_json, fetch_key, days, limit_count, label, tags_json, context, template, schedule_json, operator_hint_json, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( name, raw_query, search_query, db_keyword, json.dumps(exclude_words, ensure_ascii=False), fetch_key, days, limit, label, json.dumps(_normalize_tags(tags), ensure_ascii=False), context, template, json.dumps(schedule or {}, ensure_ascii=False), json.dumps(operator_hints or {}, ensure_ascii=False), now, now, ), ) return get_rule(name) def get_rule(name_or_id: str | int) -> Dict[str, Any]: query = "SELECT id, name, raw_query, search_query, db_keyword, exclude_json, fetch_key, days, limit_count, label, tags_json, context, template, schedule_json, operator_hint_json, created_at, updated_at, last_checked_at, last_new_count FROM watch_rules WHERE {} = ?" field = "id" if isinstance(name_or_id, int) or str(name_or_id).isdigit() else "name" with connect() as conn: row = conn.execute(query.format(field), (int(name_or_id) if field == "id" else name_or_id,)).fetchone() if not row: raise KeyError(f"watch rule not found: {name_or_id}") return { "id": row[0], "name": row[1], "raw_query": row[2], "search_query": row[3], "db_keyword": row[4], "exclude_words": json.loads(row[5]), "fetch_key": row[6], "days": row[7], "limit": row[8], "label": row[9], "tags": json.loads(row[10] or "[]"), "context": row[11], "template": row[12], "schedule": json.loads(row[13] or "{}"), "operator_hints": json.loads(row[14] or "{}"), "created_at": row[15], "updated_at": row[16], "last_checked_at": row[17], "last_new_count": row[18], } def remove_rule(name_or_id: str | int) -> int: field = "id" if isinstance(name_or_id, int) or str(name_or_id).isdigit() else "name" with connect() as conn: cur = conn.execute(f"DELETE FROM watch_rules WHERE {field} = ?", (int(name_or_id) if field == 'id' else name_or_id,)) return int(cur.rowcount or 0) def mark_seen(watch_id: int, items: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: now = datetime.now().isoformat(timespec="seconds") new_items: List[Dict[str, Any]] = [] with connect() as conn: for item in items: link = str(item.get("link", "") or "").strip() if not link: continue exists = conn.execute("SELECT 1 FROM seen_items WHERE watch_id = ? AND link = ?", (watch_id, link)).fetchone() if exists: continue conn.execute( "INSERT INTO seen_items(watch_id, link, published_at, first_seen_at) VALUES (?, ?, ?, ?)", (watch_id, link, item.get("pub_date_iso"), now), ) new_items.append(item) conn.execute( "UPDATE watch_rules SET last_checked_at = ?, last_new_count = ? WHERE id = ?", (now, len(new_items), watch_id), ) return new_items FILE:scripts/_paths.py from __future__ import annotations from pathlib import Path SKILL_DIR = Path(__file__).resolve().parent.parent DATA_DIR = SKILL_DIR / "data" CONFIG_PATH = DATA_DIR / "config.json" DB_PATH = DATA_DIR / "watch_state.db" TEST_CACHE_DIR = DATA_DIR / ".pytest-cache" UPSTREAM_REPO = Path(r"C:\Users\김태완\.openclaw\workspace\navernews-tabsearch") def ensure_data_dir() -> Path: DATA_DIR.mkdir(parents=True, exist_ok=True) return DATA_DIR

naver-news-briefing

Comments (0)