parser.py - master - fe328/pplogic

parser.py
140 строк4 KB
Алексей Кирьянов
Initial commit
29 дек 2024, 11:20
29 дек 2024, 11:20c66c2f1
import tokenize as t
import calculator as calc
from typing import List, Callable, Optional
from datetime import datetime, timedelta
from constants import ALLOWED_TOKENS, SKIPPING_TOKENS
from shunting_yard import shunting_yard

# class Expr():
#     def __init__(self, in_list:List[t.TokenInfo], *args, **kwargs):
#         self.source = in_list.copy()

#     def process(self, context):
#         pass

#     @staticmethod
#     def from_list( in_list:List[t.TokenInfo] ):
#         source = in_list.copy()
#         # if source[0].exact_type == 

# class TokenOp(Expr):
#     pass

# class OpPlus(TokenOp):
#     pass
# class OpMinus(TokenOp):
#     pass

# class Func(Expr):
#     name:str
#     param_count:int

#     def __init__(self, *vars):
#         self._vars = vars.copy()
#         if len( self._vars ) != self.param_count:
#             raise ValueError(f"{self.name} accepts exactly {self.param_count} params")

# class FuncTimeLeft(Func):
#     name = "time_left"
#     param_count = 1

#     def __init__(self, *vars):
#         super().__init__(*vars)
#         self._var = self._vars[0]
#         if not isinstance( self._var, Expr ):
#             raise ValueError("Argument must be Expr")
#         self._var:Expr
    
#     def compiler(self)->Callable:
#         def _compiler(context)->int:
#             value = self._var._process(context)
#             if not isinstance(value, datetime):
#                 raise ValueError("Argument must be datetime")
#             value:datetime
#             return (value - datetime.now()).total_seconds
#         return _compiler

# class FuncTimePast(Func):
#     name = "time_past"
#     param_count = 1

#     def __init__(self, *vars):
#         super().__init__(*vars)
#         self._var = self._vars[0]
#         if not isinstance( self._var, Expr ):
#             raise ValueError("Argument must be Expr")
#         self._var:Expr
    
#         def _process(context)->int:
#             value = self._var._process(context)
#             if not isinstance(value, datetime):
#                 raise ValueError("Argument must be datetime")
#             value:datetime
#             return (datetime.now()-value).total_seconds
#         return _compiler



# # ALLOWED_FUNCTIONS = {
# #     'time_left': (1, time_left)
# # }



def process_live_list( line:List[t.TokenInfo], context:Optional[dict] = None):
    res = calc.compute_expression(shunting_yard(line), context=context)
    print( bool(res), ' | ', ' '.join([i.string for i in line]) )
    return res

def process_filter_list( line:List[t.TokenInfo], context:Optional[dict] = None):
    res = calc.compute_filter(shunting_yard(line), context=context)
    print( res, ' | ', ' '.join([i.string for i in line]) )
    return res

if __name__ == "__main__":
    with open('test.py', 'rb') as f:
        context = {
            'old':{
                'ARG1': 2.54,
                'ARG2': "forevar",
                'arg3': datetime.now()+timedelta(hours=1)
            },
            'new':{
                'ARG1': 2,
                'ARG2': None,
                'arg3': datetime.now()-timedelta(hours=1)
            }
        }
        current_line = list()
        tokens = t.tokenize(f.readline)
        for token in tokens:
            if token.type == t.NEWLINE:
                process_live_list(current_line, context)
                current_line = list()
            elif token.exact_type in SKIPPING_TOKENS:
                continue
            elif token.exact_type not in ALLOWED_TOKENS:
                raise ValueError(f'Unsupported token {t.tok_name[token.exact_type]}')
            else:
                current_line.append( token )

    print('-'*30)

    with open('test2.py', 'rb') as f:
        context = {
            'ARG1': str,
            'ARG2': str,
            'arg3': datetime
        }
        current_line = list()
        tokens = t.tokenize(f.readline)
        for token in tokens:
            if token.type == t.NEWLINE:
                process_filter_list(current_line, context)
                current_line = list()
            elif token.exact_type in SKIPPING_TOKENS:
                continue
            elif token.exact_type not in ALLOWED_TOKENS:
                raise ValueError(f'Unsupported token {t.tok_name[token.exact_type]}')
            else:
                current_line.append( token )