KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use process_tag method in tox

Best Python code snippet using tox_python

ExtractController.py

Source:ExtractController.py

...53        del md_text54        # å¼å§å¤çä¸å¯ååºå55        md_dict, unmd_dict = self.processor_dict["symmetry_block_processor"].process_immutable_tag(md_dict, unmd_dict)56        # å¼å§å¤çå·¦ä¾§åºå ï¼åæ¬ åºåå¼ç¨æ¨¡å å åè¡¨å¼ç¨æ¨¡åï¼57        md_dict, unmd_dict = self.processor_dict["block_quotes_processor"].process_tag(md_dict, unmd_dict)58        md_dict, unmd_dict = self.processor_dict["symmetry_inline_processor"].process_immutable_tag(md_dict, unmd_dict)59        # å¼å§å¤çåè¡¨åºå60        md_dict = self.processor_dict["list_processor"].process_tag(md_dict)61        # audio/video/image tag/quotation/link htmlï¼é¿åä¹åçå²çªï¼/ å¨å¯¹ç§°åºåä¹åé¿åLinkä¸æåè¢«å¤ç62        md_dict, ref_dict = self.processor_dict["media_processor"].get_all_reference(md_dict)63        md_dict = self.processor_dict["image_processor"].process_tag(md_dict, ref_dict)64        md_dict = self.processor_dict["audio_processor"].process_tag(md_dict, ref_dict)65        md_dict = self.processor_dict["video_processor"].process_tag(md_dict, ref_dict)66        md_dict = self.processor_dict["link_processor"].process_tag(md_dict, ref_dict)67        del ref_dict68        # å¼å§å¤çå¯¹ç§°åºåï¼åé¢éè¦å®¹å¿åè¡¨ï¼69        # TODO æ¹ä¸ºæ éå®¹å¿åè¡¨70        md_dict = self.processor_dict["symmetry_block_processor"].process_mutable_tag(md_dict)71        # å¼å§å¤çè¡¨æ ¼åºåï¼æ éå®¹å¿åè¡¨ï¼72        md_dict = self.processor_dict["table_block_processor"].process_tag(md_dict)73        # å¼å§å¤çinlineçåºåä»¬74        md_dict = self.processor_dict["header_processor"].process_tag(md_dict)75        md_dict = self.processor_dict["horizontal_rule_processor"].process_tag(md_dict)76        md_dict, citations_dict = self.processor_dict["citation_processor"].process_tag(md_dict)77        md_dict = self.processor_dict["symmetry_inline_processor"].process_mutable_tag(md_dict)78        md_dict = self.processor_dict["paragraph_processor"].process_tag(md_dict)79        md_dict = self.processor_dict["break_line_processor"].process_tag(md_dict)...

collect_resume.py

Source:collect_resume.py

1# -*- coding:utf-8 _-*-2# !/usr/bin/python3import psycopg24from data_structure.csv_to_postgre import getCon5# å¤çæ¶é´ï¼è¿åå¼å§æ¶é´ï¼ç»ææ¶é´ï¼æ¶é´é¿åº¦6import re7def get_split_time(re_time):8    return re.findall('\d+', re_time)9def process_time(resume_time):10    split_time = get_split_time(resume_time)11    sp_start_time = [0, 1, 1]  # å¹´ãæãæ¥12    sp_end_time = [0, 1, 1]13    process_tag = 014    for item1 in split_time:15        item = item116        if process_tag > 5:17            break18        if len(item) > 4:19            item = item[0:3]20        if len(item) == 4:21            if process_tag == 0:22                sp_start_time[0] = int(item)23                process_tag = 124            elif process_tag <= 3:25                sp_end_time[0] = int(item)26                process_tag = 427            else:28                break29        else:30            if process_tag == 1 or process_tag == 2:31                sp_start_time[process_tag] = int(item)32                process_tag += 133            elif process_tag == 4 or process_tag == 5:34                sp_end_time[process_tag - 3] = int(item)35                process_tag += 136    time_length = -137    if sp_start_time[0] == 0 or sp_end_time[0] == 0:38        pass39    else:40        time_length = sp_end_time[0] - sp_start_time[0]41    if sp_start_time[1] > 12:42        sp_start_time[1] = 143        sp_start_time[2] = 144    if sp_end_time[1] > 12:45        sp_end_time[1] = 146        sp_end_time[2] = 147    if sp_start_time[2] > 31:48        sp_start_time[2] = 149    if sp_end_time[2] > 31:50        sp_end_time[2] = 151    return sp_start_time, sp_end_time, time_length52if __name__ == '__main__':53    Connection = getCon(database='cof', user='postgres', password='postgres', host='192.168.10.6')54    select_sql = "SELECT id_index, time_and_job, message_source FROM crawler.officer_message WHERE head_image = '';"55    insert_sql = "INSERT INTO crawler.officer_resume (id_index, id_index_n, start_time, end_time, resume, " \56                 "message_source, work_age) VALUES ('{0}', '{1}', '{2}', '{3}'" \57                 ", '{4}', '{5}', '{6}') "58    cur = Connection.cursor()59    cur.execute(select_sql)60    for line in cur.fetchall():61        id_index, time_and_job, message_source = line62        print('å¼å§å¾æ°æ®åºåå¥' + id_index + 'çå±¥åä¿¡æ¯')63        split_time_job = time_and_job.split(', ')64        # print(split_time_job, len(split_time_job))65        # try:66        for i in range(int((len(split_time_job) + 1) / 2)):67            time = split_time_job[i * 2]68            try:69                resume = split_time_job[i * 2 + 1]70            except IndexError:71                resume = ''72            id_index_n = id_index + '_' + str(i + 1)73            start_time, end_time, work_age = process_time(time)74            new_start_time = re.sub('\[|\]', '', str(start_time)).replace(', ', '-')75            new_end_time = re.sub('\[|\]', '', str(end_time)).replace(', ', '-')76            finish_insert_sql = insert_sql.format(id_index, id_index_n, new_start_time, new_end_time, resume,77                                                  message_source, work_age)78            print(finish_insert_sql)79            try:80                cur.execute(finish_insert_sql)81            except psycopg2.IntegrityError:82                print(id_index + ' çç¬¬' + str(i + 1) + ' æ¡å±¥å å·²**åå¨**æ°æ®åº')83                Connection.rollback()84                continue85            except psycopg2.DataError:86                print(id_index + ' çç¬¬' + str(i + 1) + ' æ¡å±¥å &&&åºé&&&')87                Connection.rollback()88                continue89            Connection.commit()90            print(id_index + ' çç¬¬' + str(i + 1) + ' æ¡å±¥å å·²åå¥æ°æ®åº')91        # except IndexError:92        #     None...

log.py

Source:log.py

1import logging2import logging.handlers3import os4import sys5from termcolor import colored6from datetime import datetime7import aj8LOG_DIR = '/var/log/ajenti'9LOG_NAME = 'ajenti.log'10LOG_FILE = os.path.join(LOG_DIR, LOG_NAME)11LOG_PARAMS = {12    'master_pid': None,13    'tag': 'master',14}15class ConsoleHandler(logging.StreamHandler):16    def __init__(self, stream):17        logging.StreamHandler.__init__(self, stream)18    def handle(self, record):19        if not self.stream.isatty():20            return logging.StreamHandler.handle(self, record)21        s = ''22        d = datetime.fromtimestamp(record.created)23        s += colored(d.strftime("%d.%m.%Y %H:%M  "), 'white')24        process_tag = ''25        padding = ''26        if LOG_PARAMS['tag'] == 'master':27            if os.getpid() == LOG_PARAMS['master_pid']:28                process_tag = colored('master', 'yellow')29            else:30                LOG_PARAMS['tag'] = None31                # process_tag = colored('...   ', 'yellow')32        if LOG_PARAMS['tag'] == 'restricted':33            process_tag = colored('rstrct', 'red')34            padding = '  '35        if LOG_PARAMS['tag'] == 'worker':36            process_tag = colored('worker', 'green')37            padding = '  '38        if LOG_PARAMS['tag'] == 'task':39            process_tag = colored('task  ', 'blue')40            padding = '    '41        if LOG_PARAMS['tag'] is None:42            process_tag = colored('...   ', 'blue')43            padding = '  '44        s += colored('[', 'white')45        s += process_tag46        s += colored(' %5i]  ' % os.getpid(), 'white')47        if aj.debug:48            s += colored(49                ('%15s:%-4s  ' % (record.filename[-15:], record.lineno)),50                'grey',51                attrs=['bold']52            )53        l = ''54        if record.levelname == 'DEBUG':55            l = colored('DEBUG', 'white')56        if record.levelname == 'INFO':57            l = colored('INFO ', 'green', attrs=['bold'])58        if record.levelname == 'WARNING':59            l = colored('WARN ', 'yellow', attrs=['bold'])60        if record.levelname == 'ERROR':61            l = colored('ERROR', 'red', attrs=['bold'])62        s += l + '  '63        s += padding64        s += record.msg % record.args65        s += '\n'66        self.stream.write(s)67def init_console(log_level=logging.INFO):68    log = logging.getLogger()69    log.setLevel(logging.DEBUG)70    stdout = ConsoleHandler(sys.stdout)71    stdout.setLevel(log_level)72    dformatter = logging.Formatter('%(asctime)s %(levelname)-8s [%(process)-5d]: %(message)s')73    stdout.setFormatter(dformatter)74    log.handlers = [stdout]75def init_log_directory():76    if not os.path.exists(LOG_DIR):77        os.mkdir(LOG_DIR)78def init_log_rotation():79    log = logging.getLogger()80    try:81        handler = logging.handlers.TimedRotatingFileHandler(82            os.path.join(LOG_DIR, LOG_NAME),83            when='midnight',84            backupCount=785        )86        handler.setLevel(logging.INFO)87        handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)-8s: %(message)s'))88        log.addHandler(handler)89    except IOError:90        pass91    return log92def set_log_params(**kwargs):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.