禅道自动化操作

task

补齐禅道日志，每日新增工时不小于8不超过12

思路

1. 登录禅道获取session
2. 取出历史1000条工时记录作为填写内容随机抽取/取出三年内所有工时记录作为填写内容随机抽取
3. 获取当前登录用户所有task和project，随机抽取一组
4. 获取指定月份工作日日期，需要去掉周六日和传统节假日
5. 需要具备新增/删除日志功能（整月）

实现

# coding:utf-8
import requests
import re
import hashlib
import datetime
import chinese_calendar
from bs4 import BeautifulSoup
import random
import logging
import os.path
import time
from colorama import Fore, Style
import sys
# logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s',level=logging.INFO)

# 使用前修改禅道账号密码
account = ''
password = ''


class Chandao():
    '''
    禅道相关
    '''
    def __init__(self, account, password, year, month):
        self.account = account
        self.password = password
        self.year = year
        self.month = month
        self.session = self.login()

    def login(self):
        '''
        登录禅道
        :return: session
        '''
        s = requests.Session()
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.100.4844.11 Safari/537.36"
        }
        while (True):
            rs1 = s.get("https://chandao.tingyun.com/pro/user-login.html", headers=headers)
            rs1.encoding = 'utf-8'
            rand = re.findall(r"'verifyRand' value='(.+?)'", rs1.text)
            if len(rand[0]) == 10:
                vrand = rand[0]
                break
        hash = hashlib.md5()
        hash.update(password.encode('utf-8'))
        f = hash.hexdigest() + vrand
        hash2 = hashlib.md5(f.encode('utf-8'))
        pwd = hash2.hexdigest()
        data = {
            "account": account,
            "password": pwd,
            "verifyRand": vrand
        }
        rs2 = s.post("https://chandao.tingyun.com/pro/user-login.html", headers=headers, data=data)
        rs2.encoding = 'utf-8'
        res4 = s.get('https://chandao.tingyun.com/pro/qa/')
        result = re.findall(r"\<a href=\'\/pro\/user-logout.html' \>(.+?)\<\/a\>", res4.text)
        if result[0] == "退出":
            Logger().info(f"Account:{account}————> Login Success!")
        return s

    def get_task(self):
        '''
        返回当前登录用户的所有task和project
        :return: task，project
        '''
        projects = []
        task = []
        res = self.session.get('https://chandao.tingyun.com/pro/effort-batchCreate.html?onlybody=yes')
        # print(res.text)
        # result = set(re.findall(r"<option value='task(.+?)data-keys", res.text))
        # project = set(re.findall(f"<option value='(.+?)' title=(.+?)data-keys",res.text))
        soup = BeautifulSoup(res.text, features="lxml")
        options = set(soup.findAll('option'))
        # print(len(options))
        for option in options:
            if "task" in option.get("data-keys"):
                task.append(option.get("title").split(":")[0])
            else:
                # 去除custom
                if option.get("value").isdigit():
                    # 去除0
                    if int(option.get("value"))!=0:
                        projects.append(option.get("value"))
        Logger().info("当前用户所有task："+str(task)+"\n当前用户所有project："+str(projects))
        return task,projects

    def get_current_user_history_task(self):
        history_works = []
        res = self.session.get('https://chandao.tingyun.com/pro/my-effort-all-date_desc-1000-1000-1.html')
        # print(res.text)
        soup = BeautifulSoup(res.text, features="lxml")
        for title in soup.findAll("td"):
            # 筛选clas为c-work的td标签
            if 'c-work' in title.get("class"):
                # 去掉None
                if title.get("title") != None:
                    history_works.append(title.get("title"))
        Logger().info("当前用户前1000条历史日志记录："+str(history_works))
        return history_works

    def get_current_user_annual_task(self):
        '''
        获取三年内所有日志内容
        :return:三年内日志记录的工作内容列表
        '''
        year = datetime.datetime.now().year
        annual_works = []
        for i in range(3):
            res = self.session.get(f'https://chandao.tingyun.com/pro/effort-ajaxGetEfforts-lijiacheng-{int(year)-i}.html')
            if res.status_code == 200:
                for daliy in res.json():
                    if '[T]' in daliy['title']:
                        annual_works.append(daliy['title'].split('[T]')[1])
                    else:
                        annual_works.append(daliy['title'])
            else:
                Logger().error(res.text)
        Logger().info(f"{self.account} 三年内所有日志:"+str(annual_works))
        return annual_works

    def add_daily(self):
        # 随机获取工时，保证每天的任务工时不同
        work_hours = [1, 2, 3, 4, ]
        # 当天的总工时在8～12之间
        currentday_hour = 0
        history_works = self.get_current_user_history_task()
        three_year_works = self.get_current_user_annual_task()
        cal, cal2 = Workday(self.year,self.month).getDateByTime()
        i = 0
        for c in cal:
            url = f'https://chandao.tingyun.com/pro/effort-batchCreate-{cal2[i]}.html?onlybody=yes'
            headers = {
                "content-type": "application/x-www-form-urlencoded",
                "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.100.4844.11 Safari/537.36"
            }
            while currentday_hour < 8:
                current_hour = random.choice(work_hours)
                data = {
                    "date": f"{cal[i]}",
                    "id[]": 2,
                    "work[]": f"{random.choice(history_works)}",
                    # "objectType[]": f"{random.choice(task)}",
                    # "project[]": int(random.choice(project)),
                    "left[2]": 99,
                    "consumed[]": current_hour,
                }
                # print(data)
                self.session.post(url=url, headers=headers, data=data)
                currentday_hour += current_hour
                if currentday_hour >= 8:
                    currentday_hour = 0
                    break
            i = i + 1
        Logger().info(f"{self.year}年{self.month}月日志新增完成")

    def del_daily(self):
        # 当前月份内所有可删除日志地址
        delete_url=[]
        cal, cal2 = Workday(self.year,self.month).getDateByTime()
        for c in cal2:
            url = f'https://chandao.tingyun.com/pro/my-effort-{c}.html'
            headers = {
                "content-type": "application/x-www-form-urlencoded",
                "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.100.4844.11 Safari/537.36"
            }
            res = self.session.get(url=url, headers=headers)
            soup = BeautifulSoup(res.text, features="lxml")
            options = set(soup.findAll('a'))
            for option in options:
                # print(option)
                if option.get("href") != None:
                    if "delete" in option.get("href"):
                        # url末尾增加yes，直接删除
                        href = option.get("href").split('.')
                        delete_url.append("https://chandao.tingyun.com"+href[0]+"-yes."+href[1])
        # print(delete_url)
        # 删除日志
        for url in delete_url:
            try:
                # print(url)
                r = self.session.get(url)
            except Exception as e:
                Logger().error(e)
        Logger().info(f"{self.year}年{self.month}月日志删除完成")

class Workday():
    '''
    获取填写日报的日期，规则：把当月所有的工作日时间全部返回
    '''
    def __init__(self,year, month):
        self.year = year
        self.month = month

    def getDateByTime(self):
        '''
        获取某一年某一月的工作日
        :return: %Y-%m-%d、%Y%m%d两种格式日期
        '''
        global tmp
        holidays = self.get_holidays(year=self.year,include_weekends=False)
        self.myDate = []
        self.myDate2 = []
        t = str(time.strftime(f'{self.year}-{self.month}-'))
        for i in range(1, 32):
            timeStr = t + str(i)
            try:
                # 字符串转换为规定格式的时间
                tmp = time.strptime(timeStr, '%Y-%m-%d')
                # 判断是否为周六、周日
                if (tmp.tm_wday != 6) and (tmp.tm_wday != 5) and time.strftime('%Y-%m-%d', tmp) not in holidays:
                    self.myDate.append(time.strftime('%Y-%m-%d', tmp))
                    self.myDate2.append(time.strftime('%Y%m%d', tmp))
            except Exception as e:
                Logger().error(e)
        if len(self.myDate) == 0 and time.strftime('%Y-%m-%d', tmp) not in holidays:
            self.myDate.append(time.strftime('%Y-%m-%d'))
            self.myDate2.append(time.strftime('%Y%m%d'))
        Logger().info(f"{self.year}年{self.month}月工作日为："+str(self.myDate))
        return self.myDate,self.myDate2

    def get_holidays(self,year=None, include_weekends=True):
        """
        获取某一年的所有节假日，默认当年
        :param year: which year
        :param include_weekends: False for excluding Saturdays and Sundays
        :return: list
        """
        holidays = []
        if not year:
            year = datetime.datetime.now().year
        else:
            year = year
        start = datetime.date(year, 1, 1)
        end = datetime.date(year, 12, 31)
        holidays_date = chinese_calendar.get_holidays(start, end, include_weekends)
        for holiday in holidays_date:
            holidays.append(holiday.strftime('%Y-%m-%d'))
        # print(holidays)
        return holidays


class Logger(object):
    def __init__(self):
        """
        指定保存日志的文件路径，日志级别，以及调用文件
        将日志存入到指定的文件中
        :param logger:  定义对应的程序模块名name，默认为root
        """

        # 创建一个logger
        self.logger = logging.getLogger(name=sys.argv[0])
        self.logger.setLevel(logging.DEBUG)  # 指定最低的日志级别 critical > error > warning > info > debug

        # 创建一个handler，用于写入日志文件
        rq = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime(time.time()))
        log_path = os.getcwd() + "/logs/"
        log_name = log_path + rq + ".log"
        #  这里进行判断，如果logger.handlers列表为空，则添加，否则，直接去写日志，解决重复打印的问题
        if not self.logger.handlers:
            # 创建一个handler，用于输出到控制台
            ch = logging.StreamHandler(sys.stdout)
            ch.setLevel(logging.DEBUG)

            # 定义handler的输出格式
            formatter = logging.Formatter(
                "%(asctime)s - %(filename)s[line:%(lineno)d] - %(name)s - %(message)s")
            ch.setFormatter(formatter)

            # 给logger添加handler
            # self.logger.addHandler(fh)
            self.logger.addHandler(ch)

    def debug(self, msg):
        """
        定义输出的颜色debug--white，info--green，warning/error/critical--red
        :param msg: 输出的log文字
        :return:
        """
        self.logger.debug(Fore.WHITE + "DEBUG - " + str(msg) + Style.RESET_ALL)

    def info(self, msg):
        self.logger.info(Fore.GREEN + "INFO - " + str(msg) + Style.RESET_ALL)

    def warning(self, msg):
        self.logger.warning(Fore.RED + "WARNING - " + str(msg) + Style.RESET_ALL)

    def error(self, msg):
        self.logger.error(Fore.RED + "ERROR - " + str(msg) + Style.RESET_ALL)

    def critical(self, msg):
        self.logger.critical(Fore.RED + "CRITICAL - " + str(msg) + Style.RESET_ALL)


if __name__ == '__main__':
    Logger().info('''
        -----------------使用说明-----------------------
            1. 修改文件顶部禅道账号及密码
            2. 参数说明详见readme.md文件
        -----------------谢谢使用-----------------------
        ''')
    if account == '':
        account = input("请输入禅道账号：")
    if password == '':
        password = input("请输入禅道密码：")
    try:
        if sys.argv[1] == "add":
            Chandao(account,password,int(sys.argv[2]),int(sys.argv[3])).add_daily() # 新增日志
        elif sys.argv[1] == "del":
            Chandao(account,password,int(sys.argv[2]),int(sys.argv[3])).del_daily() # 删除日志
        elif sys.argv[1] == 'query':
            Chandao(account, password, 2020,1).get_current_user_annual_task()
    except Exception as e:
        Logger().error(e)

使用方式

1.修改禅道账号和密码

2.新增日志

格式为：python3 chandao.py add xxxx x
例如：新增2020年1月日志
python3 chandao.py add 2020 1

3.删除日志

格式为：python3 chandao.py del xxxx x
例如：删除2020年1月日志
python3 chandao.py del 2020 1