Browse Source

feat: 实现摩点打卡工具核心功能

添加摩点打卡工具的核心功能模块,包括:
- 命令行参数解析和配置管理
- 摩点API客户端实现
- 打卡状态检查和打卡操作
- 用户认证和项目信息获取
- 日志记录和错误处理
- 主程序入口和结果输出

新增配置文件模板和环境变量示例
liuyuqi-cnb 1 month ago
parent
commit
612b90e4e8
8 changed files with 926 additions and 24 deletions
  1. 19 0
      .env.example
  2. 1 0
      .gitignore
  3. 24 0
      config.ini.example
  4. 196 2
      crawl_modian/__init__.py
  5. 27 10
      crawl_modian/api.py
  6. 447 12
      crawl_modian/modian.py
  7. 210 0
      crawl_modian/options.py
  8. 2 0
      requirements.txt

+ 19 - 0
.env.example

@@ -0,0 +1,19 @@
+# 摩点打卡工具环境变量配置文件
+# 复制此文件为 .env 并填入实际值
+
+# 认证方式一:Cookie(推荐,从浏览器登录后获取)
+# 格式: key1=value1; key2=value2
+# MODIAN_COOKIE=your_cookie_string_here
+
+# 认证方式二:Token
+# MODIAN_TOKEN=your_token_here
+
+# 认证方式三:用户名密码
+# MODIAN_USERNAME=your_username
+# MODIAN_PASSWORD=your_password
+
+# 日志文件路径(可选)
+# LOG_FILE=modian.log
+
+# 调试模式(可选,设置为 true 开启)
+# DEBUG=false

+ 1 - 0
.gitignore

@@ -0,0 +1 @@
+*.pyc

+ 24 - 0
config.ini.example

@@ -0,0 +1,24 @@
+# 摩点打卡工具配置文件
+# 复制此文件为 config.ini 并填入实际值
+
+[auth]
+# 认证方式一:Cookie(推荐,从浏览器登录后获取)
+# 格式: key1=value1; key2=value2
+# cookie = your_cookie_string_here
+
+# 认证方式二:Token
+# token = your_token_here
+
+# 认证方式三:用户名密码
+# username = your_username
+# password = your_password
+
+[settings]
+# 日志文件路径(可选)
+# log = modian.log
+
+# 调试模式(可选,设置为 true 开启)
+# debug = false
+
+# 输出格式(可选:text 或 json)
+# output = text

+ 196 - 2
crawl_modian/__init__.py

@@ -1,5 +1,199 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+"""
+@Contact :   liuyuqi.gov@msn.cn
+@Time    :   2023/05/17 21:14:01
+@License :   Copyright © 2017-2022 liuyuqi. All Rights Reserved.
+@Desc    :   摩点打卡工具主入口
+"""
 
 
+import sys
+import json
+from datetime import datetime
 
 
+from .modian import Modian
+from .options import parser_args
 
 
-def main():
-    pass
+
+__version__ = '1.1.0'
+
+
+def main(argv=None):
+    """主入口函数"""
+    try:
+        args = parser_args(argv)
+        
+        # 检查是否请求版本信息
+        if args.get('version', False):
+            print(f"crawl_modian v{__version__}")
+            sys.exit(0)
+        
+        # 获取命令
+        command = args.get('command', 'checkin')
+        output_format = args.get('output', 'text')
+        
+        # 创建摩点客户端实例
+        modian = Modian(args)
+        
+        # 根据命令执行相应操作
+        result = None
+        
+        if command == 'checkin':
+            # 执行打卡流程
+            result = modian.run()
+            
+        elif command == 'status':
+            # 查询打卡状态
+            if not modian.is_logged_in:
+                result = {
+                    'success': False,
+                    'message': '未提供认证信息',
+                    'timestamp': datetime.now().isoformat()
+                }
+            else:
+                result = modian.get_check_in_status()
+                
+        elif command == 'login':
+            # 仅登录
+            if args.get('cookie') or args.get('token'):
+                result = {
+                    'success': True,
+                    'message': '使用Cookie/Token认证成功',
+                    'timestamp': datetime.now().isoformat()
+                }
+            elif args.get('username') and args.get('password'):
+                login_success = modian.login()
+                result = {
+                    'success': login_success,
+                    'message': '登录成功' if login_success else '登录失败',
+                    'timestamp': datetime.now().isoformat()
+                }
+            else:
+                result = {
+                    'success': False,
+                    'message': '未提供认证信息(Cookie/Token或用户名密码)',
+                    'timestamp': datetime.now().isoformat()
+                }
+                
+        elif command == 'project':
+            # 查询项目信息
+            project_id = args.get('project_id')
+            if project_id:
+                result = modian.get_project_info(project_id)
+            else:
+                result = {
+                    'success': False,
+                    'message': '请提供项目ID (--project_id)',
+                    'timestamp': datetime.now().isoformat()
+                }
+        else:
+            result = {
+                'success': False,
+                'message': f'未知命令: {command}',
+                'timestamp': datetime.now().isoformat()
+            }
+        
+        # 输出结果
+        if output_format == 'json':
+            print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
+        else:
+            _print_result(result, command)
+        
+        # 根据结果设置退出码
+        if result.get('success', False):
+            sys.exit(0)
+        else:
+            sys.exit(1)
+            
+    except KeyboardInterrupt:
+        print('\n操作被用户中断')
+        sys.exit(1)
+    except Exception as e:
+        print(f'执行过程中发生错误: {str(e)}')
+        import traceback
+        traceback.print_exc()
+        sys.exit(1)
+
+
+def _print_result(result: dict, command: str):
+    """以文本格式打印结果"""
+    print("=" * 60)
+    print(f"摩点打卡工具 v{__version__}")
+    print("=" * 60)
+    print()
+    
+    # 打印状态
+    success = result.get('success', False)
+    status_str = "✓ 成功" if success else "✗ 失败"
+    print(f"执行状态: {status_str}")
+    print(f"执行消息: {result.get('message', '无')}")
+    print(f"执行时间: {result.get('timestamp', '未知')}")
+    print()
+    
+    # 打印详细信息
+    if command == 'checkin' and 'steps' in result:
+        print("-" * 60)
+        print("执行步骤详情:")
+        print("-" * 60)
+        
+        steps = result.get('steps', {})
+        
+        # 登录状态
+        if 'login' in steps:
+            login_step = steps['login']
+            login_status = "✓" if login_step.get('success', False) else "✗"
+            print(f"  [登录] {login_status} {login_step.get('message', '')}")
+        
+        # 打卡状态
+        if 'check_status' in steps:
+            status_step = steps['check_status']
+            status_msg = status_step.get('message', '')
+            if status_step.get('checked_in', False):
+                print(f"  [状态] ✓ 今日已打卡")
+                if status_step.get('continuous_days'):
+                    print(f"         连续打卡: {status_step['continuous_days']} 天")
+                if status_step.get('total_days'):
+                    print(f"         累计打卡: {status_step['total_days']} 天")
+            else:
+                print(f"  [状态] ✗ 今日未打卡")
+        
+        # 打卡操作
+        if 'check_in' in steps:
+            checkin_step = steps['check_in']
+            checkin_status = "✓" if checkin_step.get('success', False) else "✗"
+            print(f"  [打卡] {checkin_status} {checkin_step.get('message', '')}")
+        
+        # 用户信息
+        if 'user_info' in steps:
+            user_step = steps['user_info']
+            user_status = "✓" if user_step.get('success', False) else "✗"
+            print(f"  [用户] {user_status} {user_step.get('message', '')}")
+        
+        print()
+    
+    # 打印项目信息
+    if command == 'project' and 'data' in result:
+        print("-" * 60)
+        print("项目信息:")
+        print("-" * 60)
+        
+        data = result.get('data', {})
+        if isinstance(data, dict):
+            for key, value in data.items():
+                if not isinstance(value, (dict, list)):
+                    print(f"  {key}: {value}")
+        print()
+    
+    # 打印原始数据(调试用)
+    if 'data' in result and result['data']:
+        print("-" * 60)
+        print("原始响应数据:")
+        print("-" * 60)
+        data_str = json.dumps(result['data'], ensure_ascii=False, indent=2, default=str)
+        # 限制输出长度
+        if len(data_str) > 500:
+            data_str = data_str[:500] + "\n... (数据已截断)"
+        print(data_str)
+        print()
+    
+    print("=" * 60)

+ 27 - 10
crawl_modian/api.py

@@ -4,19 +4,36 @@
 @Contact :   liuyuqi.gov@msn.cn
 @Contact :   liuyuqi.gov@msn.cn
 @Time    :   2023/05/17 21:17:34
 @Time    :   2023/05/17 21:17:34
 @License :   Copyright © 2017-2022 liuyuqi. All Rights Reserved.
 @License :   Copyright © 2017-2022 liuyuqi. All Rights Reserved.
-@Desc    :   
+@Desc    :   摩点API配置
 '''
 '''
 
 
-host="https://www.modian.com"
-
-project_info_url = host + "https://zhongchou.modian.com/item"
+host = "https://www.modian.com"
+api_host = "https://zhongchou.modian.com"
+
+project_info_url = api_host + "/item"
+
+headers = {
+    "Host": "zhongchou.modian.com",
+    "Connection": "keep-alive",
+    "Cache-Control": "max-age=0",
+    "Upgrade-Insecure-Requests": "1",
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36",
+    "Accept": "application/json, text/plain, */*",
+    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+    "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
+    "X-Requested-With": "XMLHttpRequest",
+}
 
 
-headers={ "Host": "zhongchou.modian.com",
-         "Connection": "keep-alive",
-         "Cache-Control": "max-age=0",
-         "Upgrade-Insecure-Requests": "1",
-            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 \
-            (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36",
+api_endpoints = {
+    "login": host + "/login",
+    "user_info": host + "/user/info",
+    "check_in": host + "/user/checkin",
+    "check_in_status": host + "/user/checkin/status",
+    "project_list": api_host + "/item/list",
+    "project_detail": api_host + "/item/{project_id}",
+    "project_support": api_host + "/item/support",
+    "project_comments": api_host + "/item/{project_id}/comments",
 }
 }
 
 
 
 

+ 447 - 12
crawl_modian/modian.py

@@ -1,20 +1,455 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+'''
+@Contact :   liuyuqi.gov@msn.cn
+@Time    :   2023/05/17 21:17:34
+@License :   Copyright © 2017-2022 liuyuqi. All Rights Reserved.
+@Desc    :   摩点打卡核心类
+'''
+
 import requests
 import requests
+import json
+import logging
+from datetime import datetime
+from typing import Optional, Dict, Any, List
+
 from crawl_modian import api
 from crawl_modian import api
 
 
+
 class Modian(object):
 class Modian(object):
-    ''''''
-    def __init__(self):
-        self.sess= requests.Session()
+    '''摩点打卡客户端'''
+
+    def __init__(self, params: dict = None) -> None:
+        '''初始化摩点客户端
+        
+        Args:
+            params: 参数字典,可能包含:
+                - username: 用户名
+                - password: 密码
+                - token: 登录令牌
+                - cookie: Cookie字符串
+                - log: 日志文件路径
+                - debug: 是否开启调试模式
+        '''
+        self.params = params or {}
+        self.sess = requests.Session()
         self.sess.headers.update(api.headers)
         self.sess.headers.update(api.headers)
         
         
-    def run(self):
-        pass
+        # 配置日志
+        self._setup_logging()
+        
+        # 初始化登录状态
+        self.is_logged_in = False
+        self.user_info = None
+        
+        # 尝试使用已有的认证信息
+        self._init_auth()
+
+    def _setup_logging(self) -> None:
+        '''配置日志'''
+        log_level = logging.DEBUG if self.params.get('debug', False) else logging.INFO
+        log_file = self.params.get('log', None)
+        
+        logging.basicConfig(
+            filename=log_file,
+            level=log_level,
+            format='%(asctime)s %(levelname)s %(message)s',
+            datefmt='%Y-%m-%d %H:%M:%S'
+        )
+        self.logger = logging.getLogger(__name__)
+
+    def _init_auth(self) -> None:
+        '''初始化认证信息'''
+        # 检查是否提供了 cookie
+        cookie = self.params.get('cookie', None)
+        if cookie:
+            self._set_cookie(cookie)
+            self.is_logged_in = True
+            self.logger.info("使用提供的Cookie进行认证")
+        
+        # 检查是否提供了 token
+        token = self.params.get('token', None)
+        if token:
+            self.sess.headers.update({'Authorization': f'Bearer {token}'})
+            self.is_logged_in = True
+            self.logger.info("使用提供的Token进行认证")
+
+    def _set_cookie(self, cookie_str: str) -> None:
+        '''设置Cookie
+        
+        Args:
+            cookie_str: Cookie字符串,格式如 "key1=value1; key2=value2"
+        '''
+        cookie_dict = {}
+        for item in cookie_str.split(';'):
+            if '=' in item:
+                key, value = item.strip().split('=', 1)
+                cookie_dict[key.strip()] = value.strip()
+        
+        self.sess.cookies.update(cookie_dict)
+        self.logger.debug("Cookie已设置")
 
 
-    def get_project_info(self, project_id):
-        '''get project info'''
-        self.sess.get(api.project_info_url+"/{}.html".format(project_id)
-        , headers=api.headers)
+    def login(self, username: str = None, password: str = None) -> bool:
+        '''登录摩点账号
+        
+        Args:
+            username: 用户名/手机号/邮箱
+            password: 密码
+            
+        Returns:
+            bool: 登录是否成功
+        '''
+        username = username or self.params.get('username')
+        password = password or self.params.get('password')
+        
+        if not username or not password:
+            self.logger.error("用户名或密码为空")
+            return False
+        
+        self.logger.info(f"尝试登录账号: {username}")
+        
+        # 这里需要根据摩点实际的登录API来实现
+        # 由于摩点登录可能涉及验证码等复杂机制,
+        # 我们提供一个基础框架,实际使用时可能需要抓包分析
+        
+        login_url = api.api_endpoints['login']
+        
+        try:
+            # 先访问登录页面获取必要的token
+            response = self.sess.get(login_url)
+            response.raise_for_status()
+            
+            # 这里可能需要从页面中提取_csrf token等
+            # 实际实现时需要根据摩点的具体登录机制调整
+            
+            login_data = {
+                'username': username,
+                'password': password,
+                # 可能需要添加其他字段如 _csrf, captcha 等
+            }
+            
+            response = self.sess.post(login_url, data=login_data)
+            response.raise_for_status()
+            
+            # 检查登录是否成功
+            # 这里需要根据实际返回结果判断
+            if response.status_code == 200:
+                self.is_logged_in = True
+                self.logger.info("登录成功")
+                return True
+            else:
+                self.logger.error(f"登录失败,状态码: {response.status_code}")
+                return False
+                
+        except Exception as e:
+            self.logger.error(f"登录过程中发生错误: {str(e)}")
+            return False
 
 
-    def get_project_rank(self, project_id):
-        '''get project rank'''
-        pass
+    def check_in(self) -> Dict[str, Any]:
+        '''执行打卡操作
+        
+        Returns:
+            Dict: 打卡结果,包含成功状态和消息
+        '''
+        if not self.is_logged_in:
+            self.logger.warning("未登录,无法执行打卡")
+            return {
+                'success': False,
+                'message': '未登录,请先登录',
+                'timestamp': datetime.now().isoformat()
+            }
+        
+        self.logger.info("开始执行打卡操作")
+        
+        check_in_url = api.api_endpoints['check_in']
+        
+        try:
+            response = self.sess.post(check_in_url)
+            response.raise_for_status()
+            
+            # 尝试解析JSON响应
+            try:
+                result = response.json()
+                self.logger.info(f"打卡响应: {json.dumps(result, ensure_ascii=False)}")
+                
+                # 根据实际返回结构判断是否成功
+                success = result.get('success', False) or result.get('code', 0) == 0
+                message = result.get('message', '打卡完成')
+                
+                return {
+                    'success': success,
+                    'message': message,
+                    'data': result,
+                    'timestamp': datetime.now().isoformat()
+                }
+            except json.JSONDecodeError:
+                # 如果不是JSON响应,检查响应内容
+                content = response.text
+                self.logger.debug(f"打卡响应内容: {content[:200]}")
+                
+                # 根据实际情况判断
+                if '成功' in content or 'success' in content.lower():
+                    return {
+                        'success': True,
+                        'message': '打卡成功',
+                        'data': {'raw_response': content},
+                        'timestamp': datetime.now().isoformat()
+                    }
+                else:
+                    return {
+                        'success': False,
+                        'message': '打卡失败,请检查响应',
+                        'data': {'raw_response': content},
+                        'timestamp': datetime.now().isoformat()
+                    }
+                
+        except Exception as e:
+            self.logger.error(f"打卡过程中发生错误: {str(e)}")
+            return {
+                'success': False,
+                'message': f'打卡异常: {str(e)}',
+                'timestamp': datetime.now().isoformat()
+            }
+
+    def get_check_in_status(self) -> Dict[str, Any]:
+        '''获取打卡状态
+        
+        Returns:
+            Dict: 打卡状态信息
+        '''
+        if not self.is_logged_in:
+            self.logger.warning("未登录,无法获取打卡状态")
+            return {
+                'success': False,
+                'message': '未登录,请先登录',
+                'checked_in': False,
+                'timestamp': datetime.now().isoformat()
+            }
+        
+        self.logger.info("获取打卡状态")
+        
+        status_url = api.api_endpoints['check_in_status']
+        
+        try:
+            response = self.sess.get(status_url)
+            response.raise_for_status()
+            
+            try:
+                result = response.json()
+                self.logger.info(f"打卡状态: {json.dumps(result, ensure_ascii=False)}")
+                
+                return {
+                    'success': True,
+                    'message': '获取状态成功',
+                    'data': result,
+                    'checked_in': result.get('checked_in', False) or result.get('todayChecked', False),
+                    'continuous_days': result.get('continuous_days', 0) or result.get('continuousCheckInDays', 0),
+                    'total_days': result.get('total_days', 0) or result.get('totalCheckInDays', 0),
+                    'timestamp': datetime.now().isoformat()
+                }
+            except json.JSONDecodeError:
+                content = response.text
+                self.logger.debug(f"状态响应内容: {content[:200]}")
+                
+                return {
+                    'success': True,
+                    'message': '获取状态成功',
+                    'data': {'raw_response': content},
+                    'checked_in': '已打卡' in content or 'checked' in content.lower(),
+                    'timestamp': datetime.now().isoformat()
+                }
+                
+        except Exception as e:
+            self.logger.error(f"获取打卡状态时发生错误: {str(e)}")
+            return {
+                'success': False,
+                'message': f'获取状态异常: {str(e)}',
+                'checked_in': False,
+                'timestamp': datetime.now().isoformat()
+            }
+
+    def get_user_info(self) -> Dict[str, Any]:
+        '''获取用户信息
+        
+        Returns:
+            Dict: 用户信息
+        '''
+        if not self.is_logged_in:
+            self.logger.warning("未登录,无法获取用户信息")
+            return {
+                'success': False,
+                'message': '未登录,请先登录',
+                'timestamp': datetime.now().isoformat()
+            }
+        
+        self.logger.info("获取用户信息")
+        
+        user_info_url = api.api_endpoints['user_info']
+        
+        try:
+            response = self.sess.get(user_info_url)
+            response.raise_for_status()
+            
+            try:
+                result = response.json()
+                self.user_info = result
+                self.logger.info(f"用户信息获取成功")
+                
+                return {
+                    'success': True,
+                    'message': '获取用户信息成功',
+                    'data': result,
+                    'timestamp': datetime.now().isoformat()
+                }
+            except json.JSONDecodeError:
+                content = response.text
+                return {
+                    'success': True,
+                    'message': '获取用户信息成功',
+                    'data': {'raw_response': content},
+                    'timestamp': datetime.now().isoformat()
+                }
+                
+        except Exception as e:
+            self.logger.error(f"获取用户信息时发生错误: {str(e)}")
+            return {
+                'success': False,
+                'message': f'获取用户信息异常: {str(e)}',
+                'timestamp': datetime.now().isoformat()
+            }
+
+    def get_project_info(self, project_id: int) -> Dict[str, Any]:
+        '''获取项目信息
+        
+        Args:
+            project_id: 项目ID
+            
+        Returns:
+            Dict: 项目信息
+        '''
+        self.logger.info(f"获取项目信息: {project_id}")
+        
+        project_url = api.api_endpoints['project_detail'].format(project_id=project_id)
+        
+        try:
+            response = self.sess.get(project_url)
+            response.raise_for_status()
+            
+            try:
+                result = response.json()
+                return {
+                    'success': True,
+                    'message': '获取项目信息成功',
+                    'data': result,
+                    'timestamp': datetime.now().isoformat()
+                }
+            except json.JSONDecodeError:
+                content = response.text
+                return {
+                    'success': True,
+                    'message': '获取项目信息成功',
+                    'data': {'raw_response': content},
+                    'timestamp': datetime.now().isoformat()
+                }
+                
+        except Exception as e:
+            self.logger.error(f"获取项目信息时发生错误: {str(e)}")
+            return {
+                'success': False,
+                'message': f'获取项目信息异常: {str(e)}',
+                'timestamp': datetime.now().isoformat()
+            }
+
+    def get_project_rank(self, project_id: int) -> Dict[str, Any]:
+        '''获取项目排名/支持者列表
+        
+        Args:
+            project_id: 项目ID
+            
+        Returns:
+            Dict: 排名信息
+        '''
+        self.logger.info(f"获取项目排名: {project_id}")
+        
+        # 这里需要根据摩点实际的API来实现
+        # 排名通常在项目详情中,或者有专门的接口
+        
+        try:
+            # 先尝试获取项目详情,可能包含排名信息
+            project_info = self.get_project_info(project_id)
+            
+            if project_info['success']:
+                return {
+                    'success': True,
+                    'message': '获取项目排名成功',
+                    'data': project_info.get('data', {}),
+                    'timestamp': datetime.now().isoformat()
+                }
+            else:
+                return project_info
+                
+        except Exception as e:
+            self.logger.error(f"获取项目排名时发生错误: {str(e)}")
+            return {
+                'success': False,
+                'message': f'获取项目排名异常: {str(e)}',
+                'timestamp': datetime.now().isoformat()
+            }
+
+    def run(self) -> Dict[str, Any]:
+        '''执行主流程:登录 -> 检查打卡状态 -> 打卡
+        
+        Returns:
+            Dict: 执行结果汇总
+        '''
+        self.logger.info("开始执行摩点打卡主流程")
+        
+        results = {
+            'timestamp': datetime.now().isoformat(),
+            'steps': {}
+        }
+        
+        # 步骤1:检查登录状态
+        if not self.is_logged_in:
+            # 尝试使用用户名密码登录
+            if self.params.get('username') and self.params.get('password'):
+                login_result = self.login()
+                results['steps']['login'] = login_result
+                if not login_result:
+                    results['success'] = False
+                    results['message'] = '登录失败'
+                    return results
+            else:
+                results['success'] = False
+                results['message'] = '未提供认证信息(Cookie/Token或用户名密码)'
+                return results
+        else:
+            results['steps']['login'] = {'success': True, 'message': '已登录'}
+        
+        # 步骤2:获取打卡状态
+        status_result = self.get_check_in_status()
+        results['steps']['check_status'] = status_result
+        
+        # 步骤3:如果今天还没打卡,执行打卡
+        if not status_result.get('checked_in', False):
+            check_in_result = self.check_in()
+            results['steps']['check_in'] = check_in_result
+            
+            if check_in_result.get('success', False):
+                results['success'] = True
+                results['message'] = '打卡成功'
+            else:
+                results['success'] = False
+                results['message'] = check_in_result.get('message', '打卡失败')
+        else:
+            results['success'] = True
+            results['message'] = '今日已打卡'
+            results['steps']['check_in'] = {'success': True, 'message': '今日已打卡,无需重复打卡'}
+        
+        # 步骤4:获取用户信息(可选)
+        user_info_result = self.get_user_info()
+        results['steps']['user_info'] = user_info_result
+        
+        self.logger.info(f"打卡流程完成,结果: {json.dumps(results, ensure_ascii=False, default=str)}")
+        
+        return results

+ 210 - 0
crawl_modian/options.py

@@ -0,0 +1,210 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+"""
+@Contact :   liuyuqi.gov@msn.cn
+@Time    :   2023/05/17 21:17:34
+@License :   Copyright © 2017-2022 liuyuqi. All Rights Reserved.
+@Desc    :   命令行参数解析
+"""
+
+import argparse
+import os
+import shlex
+import dotenv
+from collections import OrderedDict
+
+
+def parser_args(overrideArguments=None):
+    """解析命令行参数"""
+
+    argparser = argparse.ArgumentParser(
+        description='摩点打卡工具 - 自动登录并执行打卡操作',
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog='''
+示例用法:
+  # 使用Cookie打卡
+  python main.py checkin --cookie "your_cookie_string"
+  
+  # 使用用户名密码登录并打卡
+  python main.py checkin --username your_username --password your_password
+  
+  # 使用配置文件
+  python main.py checkin -c config.ini
+  
+  # 仅查询打卡状态
+  python main.py status --cookie "your_cookie_string"
+  
+  # 获取项目信息
+  python main.py project --project_id 12345
+  
+  # 调试模式
+  python main.py checkin --cookie "your_cookie_string" --debug
+        '''
+    )
+    
+    # 配置文件
+    argparser.add_argument(
+        '-c', '--config', 
+        help='配置文件路径 (默认: config.ini)', 
+        default='config.ini'
+    )
+    
+    # 命令
+    argparser.add_argument(
+        'command',
+        help='执行命令: checkin(打卡), status(查询状态), login(仅登录), project(项目信息)',
+        choices=['checkin', 'status', 'login', 'project'],
+        nargs='?',
+        default='checkin'
+    )
+    
+    # 日志文件
+    argparser.add_argument(
+        '-l', '--log', 
+        help='日志文件路径', 
+        default=None
+    )
+    
+    # 调试模式
+    argparser.add_argument(
+        '-d', '--debug', 
+        help='开启调试模式', 
+        action='store_true'
+    )
+    
+    # 认证相关参数
+    auth_group = argparser.add_argument_group('认证选项')
+    auth_group.add_argument(
+        '--cookie', 
+        help='Cookie字符串 (格式: "key1=value1; key2=value2")',
+        default=None
+    )
+    auth_group.add_argument(
+        '--token', 
+        help='登录Token',
+        default=None
+    )
+    auth_group.add_argument(
+        '--username', '--user',
+        help='用户名/手机号/邮箱',
+        default=None
+    )
+    auth_group.add_argument(
+        '--password', '--pwd',
+        help='密码',
+        default=None
+    )
+    
+    # 项目相关参数
+    project_group = argparser.add_argument_group('项目选项')
+    project_group.add_argument(
+        '--project_id', '--pid',
+        help='项目ID (用于查询项目信息)',
+        type=int,
+        default=None
+    )
+    
+    # 输出格式
+    argparser.add_argument(
+        '-o', '--output',
+        help='输出格式: json, text (默认: text)',
+        choices=['json', 'text'],
+        default='text'
+    )
+    
+    # 解析参数
+    args = argparser.parse_args(overrideArguments)
+
+    # 移除 None 值
+    command_line_conf = OrderedDict(
+        {k: v for k, v in args.__dict__.items() if v is not None}
+    )
+
+    # 读取配置文件
+    system_conf = user_conf = custom_conf = OrderedDict()
+    
+    # 读取 .env 文件
+    user_conf = _read_user_conf()
+    
+    # 读取自定义配置文件
+    if args.config:
+        custom_conf = _read_custom_conf(args.config)
+
+    # 合并配置(优先级:命令行 > 自定义配置 > .env)
+    system_conf.update(user_conf)
+    system_conf.update(custom_conf)
+    system_conf.update(command_line_conf)
+    
+    return system_conf
+
+
+def _read_custom_conf(config_path: str) -> OrderedDict:
+    """读取自定义配置文件 config.ini 或 config.yaml"""
+    
+    def compat_shlex_split(s, comments=False, posix=True):
+        if isinstance(s, str):
+            s = s.encode('utf-8')
+        return list(map(lambda s: s.decode('utf-8'), shlex.split(s, comments, posix)))
+
+    # 检查文件是否存在
+    if not os.path.exists(config_path):
+        return OrderedDict()
+    
+    # 根据扩展名选择解析方式
+    ext = os.path.splitext(config_path)[1].lower()
+    
+    try:
+        if ext in ['.yaml', '.yml']:
+            import yaml
+            with open(config_path, 'r', encoding='utf-8') as f:
+                data = yaml.safe_load(f)
+                return OrderedDict(data) if data else OrderedDict()
+        elif ext in ['.json']:
+            import json
+            with open(config_path, 'r', encoding='utf-8') as f:
+                data = json.load(f)
+                return OrderedDict(data) if data else OrderedDict()
+        else:
+            # 默认使用 ini 格式或 shlex 解析
+            try:
+                import configparser
+                config = configparser.ConfigParser()
+                config.read(config_path, encoding='utf-8')
+                
+                result = OrderedDict()
+                for section in config.sections():
+                    for key, value in config.items(section):
+                        result[key] = value
+                return result
+            except Exception:
+                # 如果 ini 解析失败,尝试 shlex 解析
+                with open(config_path, 'r', encoding='utf-8') as f:
+                    contents = f.read()
+                    res = compat_shlex_split(contents, comments=True)
+                    
+                    # 将列表转换为字典
+                    result = OrderedDict()
+                    for i in range(0, len(res) - 1, 2):
+                        key = res[i].lstrip('-')
+                        if i + 1 < len(res):
+                            result[key] = res[i + 1]
+                    return result
+                    
+    except Exception as e:
+        print(f"读取配置文件时发生错误: {str(e)}")
+        return OrderedDict()
+
+
+def _read_user_conf() -> OrderedDict:
+    """读取用户配置文件: .env 文件"""
+    user_conf = OrderedDict()
+    
+    # 检查当前目录和上级目录的 .env 文件
+    env_paths = ['.env', os.path.join('..', '.env')]
+    
+    for dotenv_path in env_paths:
+        if os.path.exists(dotenv_path):
+            user_conf = dotenv.dotenv_values(dotenv_path)
+            break
+    
+    return OrderedDict(user_conf)

+ 2 - 0
requirements.txt

@@ -0,0 +1,2 @@
+requests>=2.22.0
+python-dotenv>=1.0.0