对外py编写

schoolNewsCrawler/crawler/xxqg/README_important_crawler.md

@@ -1,140 +0,0 @@
-# 学习强国重要新闻爬虫使用说明
-
-## 功能概述
-
-在 `XxqgCrawler` 类中新增了 `crawl_important` 方法，用于爬取学习强国"重要新闻"栏目的文章内容。
-
-## 实现原理
-
-该方法结合了旧版 `myQiangguo` 爬虫和新版 Selenium 爬虫的优势：
-
-1. **获取文章列表**：参考旧版爬虫方式，使用 `requests` 库直接请求 JSON 接口获取文章列表
-   - JSON接口地址: `https://www.xuexi.cn/lgdata/1jscb6pu1n2.json?_st=26095725`
-   - 返回包含文章URL、标题、来源等基础信息的列表
-
-2. **解析文章详情**：使用现有的 `parse_news_detail` 方法（基于 Selenium）解析每篇文章的详细内容
-   - 提取标题、发布时间、来源
-   - 提取正文内容（文字、图片、视频）
-   - 保存完整的文章结构
-
-## 使用方法
-
-### 基本用法
-
-```python
-from crawler.xxqg.XxqgCrawler import XxqgCrawler
-
-# 初始化爬虫
-crawler = XxqgCrawler()
-
-# 爬取重要新闻（默认最多60篇）
-result = crawler.crawl_important()
-
-# 检查结果
-if result.success:
-    print(f"成功爬取 {len(result.dataList)} 篇新闻")
-    for news in result.dataList:
-        print(f"标题: {news.title}")
-        print(f"来源: {news.source}")
-        print(f"发布时间: {news.publishTime}")
-else:
-    print(f"爬取失败: {result.message}")
-
-# 关闭浏览器
-crawler.driver.quit()
-```
-
-### 自定义爬取数量
-
-```python
-# 只爬取前10篇文章
-result = crawler.crawl_important(max_count=10)
-```
-
-### 运行测试脚本
-
-```bash
-cd f:\Project\schoolNews\schoolNewsCrawler\crawler\xxqg
-python test_important_crawler.py
-```
-
-## 输出结果
-
-爬取完成后，结果会自动保存到 `Xxqg_important_news.json` 文件中，包含以下信息：
-
-```json
-[
-    {
-        "title": "文章标题",
-        "url": "文章URL",
-        "source": "来源",
-        "publishTime": "发布时间",
-        "contentRows": [
-            {
-                "type": "text",
-                "content": "段落文本"
-            },
-            {
-                "type": "img",
-                "content": "<img src='图片URL' />"
-            }
-        ]
-    }
-]
-```
-
-## 参数说明
-
-### `crawl_important(max_count=60)`
-
-- **max_count**: 最多爬取的文章数量，默认60篇
-- **返回值**: `ResultDomain` 对象
-  - `success`: 是否成功
-  - `code`: 状态码（0表示成功，1表示失败）
-  - `message`: 提示信息
-  - `dataList`: 新闻列表（`List[NewsItem]`）
-
-## 注意事项
-
-1. **浏览器初始化**：首次运行时会自动打开 Chrome 浏览器并访问学习强国主页获取 Cookie
-2. **验证码处理**：如果遇到验证码，程序会暂停30秒让用户手动完成验证
-3. **爬取速度**：每篇文章之间会有1-2秒的随机延迟，避免请求过快被封禁
-4. **资源清理**：使用完毕后记得调用 `crawler.driver.quit()` 关闭浏览器
-
-## 与旧版爬虫的对比
-
-### 旧版爬虫 (myQiangguo)
-- 使用 `requests` + `BeautifulSoup` 解析静态HTML
-- 依赖于特定的 `data+MD5.js` 接口格式
-- 需要处理不同格式的URL（.html和.json）
-
-### 新版爬虫 (XxqgCrawler)
-- 结合 `requests` 获取列表 + `Selenium` 解析详情
-- 能够处理动态加载的内容
-- 统一的接口和返回格式
-- 更好的错误处理和日志记录
-
-## 扩展功能
-
-如果需要爬取其他栏目，可以参考 `crawl_important` 方法的实现，修改对应的 JSON 接口URL即可。
-
-常见栏目的JSON接口：
-- 重要新闻: `https://www.xuexi.cn/lgdata/1jscb6pu1n2.json?_st=26095725`
-- 重要活动: `https://www.xuexi.cn/lgdata/1jpuhp6fn73.json?_st=26095746`
-- 重要会议: `https://www.xuexi.cn/lgdata/19vhj0omh73.json?_st=26095747`
-- 重要讲话: `https://www.xuexi.cn/lgdata/132gdqo7l73.json?_st=26095749`
-
-## 技术架构
-
-```
-crawl_important()
-├── requests 获取JSON列表
-│   └── 解析文章URL和基础信息
-├── 遍历URL列表
-│   ├── parse_news_detail() (Selenium)
-│   │   ├── 访问文章页面
-│   │   ├── 提取标题、时间、来源
-│   │   └── 解析内容（文字、图片、视频）
-│   └── 补充缺失的字段
-└── 保存结果到JSON文件
-```

schoolNewsCrawler/crawler/xxqg/XxqgColumn.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+学习强国栏目爬虫命令行工具
+用法: python RmrbSearch.py --key "关键词" --total 10 --type 0
+"""
+
+import argparse
+import json
+import sys
+from pathlib import Path
+import time
+# Add project root directory to path to import crawler
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+from crawler.xxqg.XxqgCrawler import XxqgCrawler
+from loguru import logger
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(
+        description='学习强国新闻栏目爬虫工具',
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+        """
+    )
+
+    parser.add_argument(
+        '--column', '-c',
+        type=str,
+        required=True,
+        help='栏目名称 important：重要新闻， xuexishiping：学习时评，zonghexinwen：综合新闻，zhongxuanbu：中宣部'
+    )
+
+    parser.add_argument(
+        '--yesterday', '-y',
+        type=str,
+        default="True",
+        help='是否抓取昨天的数据 (默认: True)'
+    )
+
+    parser.add_argument(
+        '--start', '-s',
+        type=str,
+        default=None,
+        help='开始日期 (格式: YYYY-MM-DD)'
+    )
+
+    parser.add_argument(
+        '--end', '-e',
+        type=str,
+        default=None,
+        help='结束日期 (格式: YYYY-MM-DD)'
+    )
+
+    parser.add_argument(
+        '--output', '-o',
+        type=str,
+        help='输出文件路径'
+    )
+
+    args = parser.parse_args()
+
+    # 获取参数
+    column = args.column
+    yesterday = str(args.yesterday)
+    if yesterday.upper() == "FALSE":
+        yesterday = False
+    elif yesterday.upper() == "TRUE":
+        yesterday = True
+    else:
+        parser.error("--yesterday 参数必须是 True 或 False")
+
+    start = args.start
+    end = args.end
+    output_file = args.output
+
+    logger.info("使用直接参数模式")
+
+    # column 必须存在
+    if not column or not column.strip():
+        parser.error("栏目不能为空！")
+    try:
+        logger.info(f"开始搜索: 栏目='{column}', 昨天={yesterday}, 开始日期={start}, 结束日期={end}")
+        crawler = XxqgCrawler()
+        url_config= crawler.config.urls[column]
+        time.sleep(5)
+        result = crawler.crawl_base(url_config, yesterday=yesterday, start=start, end=end)
+        # print(result)
+        output = {
+            "code": result.code,
+            "message": result.message,
+            "success": result.success,
+            "data": None,
+            "dataList": [item.model_dump() for item in result.dataList] if result.dataList else []
+        }
+        # result = None
+        # with open("F:\Project\schoolNews\schoolNewsCrawler\output\output.json", "r", encoding="utf-8") as f:
+        #     result = json.load(f)
+        # print(result)
+        # output = result
+
+        
+        if output_file:
+            output_path = Path(output_file)
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            with open(output_path, 'w', encoding='utf-8') as f:
+                json.dump(output, f, ensure_ascii=False, indent=2)
+            logger.info(f"结果已保存到: {output_file}")
+
+        crawler.close()
+        sys.exit(0 if result.success else 1)
+        # print(json.dumps(output, ensure_ascii=False, indent=2))
+        # sys.exit(0 if result["success"] else 1)
+    except Exception as e:
+        logger.error(f"执行失败: {str(e)}")
+        error_output = {
+            "code": 500,
+            "message": f"执行失败: {str(e)}",
+            "success": False,
+            "data": None,
+            "dataList": []
+        }
+        print(json.dumps(error_output, ensure_ascii=False, indent=2))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

schoolNewsCrawler/crawler/xxqg/XxqgCrawler.py

@@ -104,7 +104,7 @@ class XxqgCrawler(BaseCrawler):
                         'sec-ch-ua-platform': '"Windows"'
                     }
                 ),
-                "zongheshiping": UrlConfig(
+                "zonghexinwen": UrlConfig(
                     url="https://www.xuexi.cn/7097477a9643eacffe4cc101e4906fdb/9a3668c13f6e303932b5e0e100fc248b.html",
                     method="GET",
                     params={
@@ -359,7 +359,7 @@ class XxqgCrawler(BaseCrawler):
                             "type": "text",
                             "content": text_content
                         })
-                        logger.debug(f"提取文字: {text_content[:50]}...")
+                        # logger.debug(f"提取文字: {text_content[:50]}...")
                 
                 except Exception as e:
                     logger.warning(f"处理内容元素失败: {str(e)}")
@@ -369,7 +369,7 @@ class XxqgCrawler(BaseCrawler):
         
         if is_page():
             pass
-        logger.info(f"解析文章详情完成: {news_item.model_dump()}")
+        logger.info(f"解析文章详情完成: {news_item.url}")
         return news_item
 
 
@@ -577,7 +577,7 @@ class XxqgCrawler(BaseCrawler):
         # 解析meta请求响应获取channelId
         try:
             meta_data = json.loads(target_request.response.body)
-            logger.info(f"Meta响应数据: {meta_data}")
+            # logger.info(f"Meta响应数据: {meta_data}")
             
             # 提取channelId
             if 'pageData' in meta_data and 'channel' in meta_data['pageData']:

schoolNewsCrawler/crawler/xxqg/XxqgSearch.py

@@ -0,0 +1,106 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+学习强国搜索爬虫命令行工具
+用法: python RmrbSearch.py --key "关键词" --total 10 --type 0
+"""
+
+import argparse
+import json
+import sys
+from pathlib import Path
+import time
+# Add project root directory to path to import crawler
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+from crawler.xxqg.XxqgCrawler import XxqgCrawler
+from loguru import logger
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(
+        description='学习强国新闻搜索工具',
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+        """
+    )
+
+    parser.add_argument(
+        '--query', '-q',
+        type=str,
+        required=True,
+        help='搜索关键词'
+    )
+
+    parser.add_argument(
+        '--total', '-t',
+        type=int,
+        default=10,
+        help='抓取数量 (默认: 10)'
+    )
+
+    parser.add_argument(
+        '--output', '-o',
+        type=str,
+        help='输出文件路径'
+    )
+
+    args = parser.parse_args()
+
+    # 获取参数
+    key = args.query
+    total = args.total
+    output_file = args.output
+
+    logger.info("使用直接参数模式")
+
+    # 关键校验：key 必须存在
+    if not key or not key.strip():
+        parser.error("搜索关键词不能为空！")
+    try:
+        logger.info(f"开始搜索: 关键词='{key}', 数量={total}")
+        crawler = XxqgCrawler()
+        time.sleep(5)
+        result = crawler.search(keyword=key.strip(), total=total)
+        # print(result)
+        output = {
+            "code": result.code,
+            "message": result.message,
+            "success": result.success,
+            "data": None,
+            "dataList": [item.model_dump() for item in result.dataList] if result.dataList else []
+        }
+        # result = None
+        # with open("F:\Project\schoolNews\schoolNewsCrawler\output\output.json", "r", encoding="utf-8") as f:
+        #     result = json.load(f)
+        # print(result)
+        # output = result
+
+        
+        if output_file:
+            output_path = Path(output_file)
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            with open(output_path, 'w', encoding='utf-8') as f:
+                json.dump(output, f, ensure_ascii=False, indent=2)
+            logger.info(f"结果已保存到: {output_file}")
+
+        crawler.close()
+        sys.exit(0 if result.success else 1)
+        # print(json.dumps(output, ensure_ascii=False, indent=2))
+        # sys.exit(0 if result["success"] else 1)
+    except Exception as e:
+        logger.error(f"执行失败: {str(e)}")
+        error_output = {
+            "code": 500,
+            "message": f"执行失败: {str(e)}",
+            "success": False,
+            "data": None,
+            "dataList": []
+        }
+        print(json.dumps(error_output, ensure_ascii=False, indent=2))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

schoolNewsCrawler/crawler/xxqg/test_important_crawler.py

@@ -1,42 +0,0 @@
-"""
-测试学习强国重要新闻爬虫
-"""
-from XxqgCrawler import XxqgCrawler
-from loguru import logger
-
-def test_crawl_important():
-    """测试爬取重要新闻"""
-    try:
-        # 初始化爬虫
-        logger.info("初始化学习强国爬虫...")
-        crawler = XxqgCrawler()
-        
-        # 爬取重要新闻（默认最多60篇）
-        logger.info("开始爬取重要新闻...")
-        result = crawler.crawl_important(max_count=10)  # 测试时只爬取10篇
-        
-        # 检查结果
-        if result.success:
-            logger.info(f"爬取成功！{result.message}")
-            logger.info(f"共爬取到 {len(result.dataList)} 篇新闻")
-            
-            # 打印前3篇新闻标题
-            for idx, news in enumerate(result.dataList[:3], 1):
-                logger.info(f"{idx}. {news.title}")
-                logger.info(f"   来源: {news.source}")
-                logger.info(f"   发布时间: {news.publishTime}")
-                logger.info(f"   内容行数: {len(news.contentRows)}")
-                logger.info("")
-        else:
-            logger.error(f"爬取失败: {result.message}")
-        
-        # 关闭浏览器
-        if crawler.driver:
-            crawler.driver.quit()
-            logger.info("浏览器已关闭")
-            
-    except Exception as e:
-        logger.exception(f"测试过程中发生错误: {str(e)}")
-
-if __name__ == "__main__":
-    test_crawl_important()