import os import time import subprocess from consultas.querys_sqlite import get_geoloc from iplocate import ( requests, token, filtro_ip_propia, selfpath, parser, log_usage, valid_ip ) from json import loads from datetime import datetime from sqlalchemy import create_engine from sqlalchemy.orm import relationship, sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Sequence, update # from sqlalchemy.orm.session import Session from sqlalchemy.sql.expression import select # , distinct from sqlalchemy.sql.schema import ForeignKey from rich.progress import Progress, track from rich.console import Console from mapsgen import maps_gen logs_dir = parser.get('bash_script', 'destino_log') logs_dir = logs_dir.strip("'") base_de_datos = f'sqlite:////{selfpath}/ipinfo.db' console = Console() Base = declarative_base() # Tabla registro ip info class Registro(Base): """Definición de tabla 'Registro'""" __tablename__ = 'registro' ip = Column(String, primary_key=True) hostname = Column(String, nullable=True) anycast = Column(String, nullable=True) cuidad = Column(String, nullable=True) region = Column(String, nullable=True) pais = Column(String, nullable=True) geoloc = Column(String, nullable=True) organizacion = Column(String, nullable=True) fecha_reg = Column(Integer, default=int(time.mktime(time.localtime()))) tzone = Column(String, nullable=True) cod_post = Column(String, nullable=True) visitas = relationship( "Visita", order_by="Visita.id", back_populates="visita_ip", cascade="all, delete, delete-orphan" ) def get_fecha(self): """Convierte fecha 'unix epoch' y devuelve en formato local""" return time.asctime( time.localtime( int(self.fecha_reg.__repr__()) ) ) def __repr__(self) -> str: try: rep = f'ip={self.ip};host={self.hostname};' \ f'anycast={self.anycast};cuidad={self.cuidad};' \ f'region={self.region};pais={self.pais};' \ f'geoloc={self.geoloc};organizacion={self.organizacion};' \ f'fecha_reg={self.get_fecha()};tzone={self.tzone};' \ f'cod_post={self.cod_post}' return rep except Exception as ex: print('Exception :', ex) return "error repr" class Visita(Base): """Definición de tabla 'Visita'""" __tablename__ = 'visita' id = Column(Integer, Sequence('visita_id_seq'), primary_key=True) ip = Column(String, ForeignKey('registro.ip')) cod_html = Column(Integer) fecha = Column(Integer) metodo = Column(String, default='---') consulta = Column(String, default='---') registro = Column(Integer, default=0) visita_ip = relationship("Registro", back_populates="visitas") def get_fecha(self): """Convierte fecha 'unix epoch' y devuelve en formato local""" return time.asctime( time.localtime( int(self.fecha.__repr__()) ) ) def consulta_registro(self): return True if self.registro == 1 else False def __repr__(self) -> str: """Representación en cadena de texto del los datos en tabla""" try: rep = f'id={self.id},ip={self.ip},html={self.cod_html},' \ f'fecha={self.get_fecha()},metodo={self.metodo}' \ f',request={self.consulta}' return rep except Exception as ex: print('Exception :', ex) return "Error repr Visita" engine = create_engine(base_de_datos) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() # Formatos fechas logs: """ # access.log == reverse_access.log # error.log == reverse_error.log fecha_error = "2022/05/10 07:11:46" fecha_access = "10/May/2022:11:42:14 -0400".split(' ')[0] """ def fecha_access_to_epoch(fecha): """Convierte la fecha del formato entregado por access.log y reverse_access.log (nginx) al formato unix epoch. :fecha: str Fecha :returns: int unix epoch fecha (secs) """ fecha = datetime.strptime(fecha, '%d/%b/%Y:%H:%M:%S') fecha_unix = int(time.mktime(fecha.timetuple())) return fecha_unix def fecha_error_to_epoch(fecha): """Convierte la fecha del formato entregado por error.log y reverse_error.log (nginx) al formato unix epoch. :fecha_local: str Fecha :returns: int unix epoch fecha (secs) """ fecha = datetime.strptime(fecha, '%Y/%m/%d %H:%M:%S') fecha_unix = int(time.mktime(fecha.timetuple())) return fecha_unix def epoch_to_local(fecha): """Convierte fecha unix epoch a localtime :fecha: int Fecha (secs) :returns: str Fecha """ return time.asctime(time.localtime(int(fecha))) def ip_registrada(ip): """Retorna respuesta a consulta valor de columna 'registro' en tabla 'Visita' para ip pasada como argumento. """ try: ip_reg = session.query(Visita).filter( Visita.ip == ip ).filter( Visita.registro == 1 ).first() except Exception as ex: print('Exception', ex) ip_reg = None return 0 if ip_reg is None else ip_reg.registro def carga_access_log(log): """Procesa logs del tipo access, filtra IPs propias (publica y locales), acorta las peticiones donde es necesario, convierte fechas a formato unix epoch, los añade a la session para la tabla 'Visita'. Finalmente realiza la transaccion utilizando clase Progres() del modulo rich. Borra el log procesado.""" if os.path.exists(log): nombre_log = log.split('/')[-1] console.print( f'[yellow]Procesando [[/yellow]{nombre_log}[yellow]][/yellow]' ) try: with open(log, 'r') as lista: try: largo = subprocess.run( ['wc', '-l', log], capture_output=True, text=True ) largo = int(largo.stdout.split(' ')[0]) for linea in track( lista, total=largo, description='[blue bold]Cargando [/blue bold]' ): ip = linea.split(' ')[0] if filtro_ip_propia(ip): try: ip = linea.split(' ')[0] except Exception as ex: ip = None print('Exception split IP', ex) try: metodo = linea.split('"')[1].split(' ')[0] if len(metodo) > 10 or len(metodo) < 2: metodo = '---' except Exception: metodo = '---' try: url = linea.split('"')[1].split(' ')[1] if len(url) > 254: url = url[:252]+'...' except Exception: url = '---' try: codigo = int(linea.split('"')[2].split(' ')[1]) if len(str(codigo)) != 3: codigo = 0 except Exception: codigo = 0 try: fecha = linea.split(' ')[3][1:] fecha = fecha_access_to_epoch(fecha) except Exception as ex: fecha = None print('Exception split Fecha:', ex) if ip_registrada(ip): session.add( Visita( ip=ip, cod_html=codigo, fecha=fecha, metodo=metodo, consulta=url, registro=1 ) ) else: session.add( Visita( ip=ip, cod_html=codigo, fecha=fecha, metodo=metodo, consulta=url ) ) except Exception as ex: print('Exception: ', ex) try: with Progress() as prog, session: task1 = prog.add_task( "[yellow bold]Guardando[/yellow bold]", total=len(session.new) ) session.commit() while not prog.finished: prog.update(task1, advance=0.1) time.sleep(0.05) except Exception as ex: print('Exception Progress: ', ex) console.print( '[magenta] - Carga completa.. borrando log[/magenta]\n' ) os.remove(log) return True except Exception: console.print( f'[red]Error al intentar abrir/cargar: [{log}[/red]]\n' ) return False else: console.print( f'[bold red]Log: [[/bold red]{log}[bold red]] ' 'inexistente.[/bold red]\n' ) return False def carga_error_logs(log): """Procesa logs del tipo error, acorta los donde es necesario, convierte fechas a formato unix epoch, filtra IPs propias (publica y locales), los añade a session para tabla 'Visita'. Finalmente realiza la transaccion utilizando clase 'Progress' del modulo rich. Borra el log procesado.""" if os.path.exists(log): nombre_log = log.split('/')[-1] console.print( f'[yellow]Procesando [[/yellow]{nombre_log}[yellow]][/yellow]' ) try: with open(log, 'r') as lista: ip, fecha, url, metodo = None, None, None, None try: largo = subprocess.run( ['wc', '-l', log], capture_output=True, text=True ) largo = int(largo.stdout.split(' ')[0]) for linea in track( lista, total=largo, description='[blue bold]Cargando [/blue bold]' ): linea = linea.split('\n')[0] if (linea.rfind('[notice]') > 0 or linea.rfind('[crit]') > 0): if linea.find('[crit]') > 0: try: ip = linea.split('client: ')[1].split(',')[0] except Exception as ex: log_usage( 'Exception Ip error_log {crit}', ex ) ip = None try: fecha = ' '.join(linea.split(' ')[0:2]) except Exception: fecha = None try: url = linea.split('"')[1].split(' ')[1] if len(url) > 254: url = url[:252]+'...' except Exception: url = ' '.join(linea.split(' ')[5:]) try: metodo = linea.split('"')[1].split(' ')[0] except Exception: metodo = '---' else: try: ip = linea.split('client: ')[1].split(',')[0] except Exception as ex: log_usage( 'Exception Ip error_log {notice}', ex ) ip = None try: fecha = ' '.join(linea.split(' ')[0:2]) except Exception: fecha = None try: metodo = linea.split('request: "')[1].split(' ')[0] except Exception: metodo = '---' try: url = linea.split('"')[1].split(' ')[0] if len(url) > 254: url = url[:252]+'...' except Exception: url = '---' if ip is not None: if filtro_ip_propia(ip): fecha = int(fecha_error_to_epoch(fecha)) codigo = 0 if ip_registrada(ip): session.add( Visita( ip=ip, cod_html=codigo, fecha=fecha, consulta=url, metodo=metodo, registro=1 ) ) else: session.add( Visita( ip=ip, cod_html=codigo, fecha=fecha, consulta=url, metodo=metodo ) ) else: log_usage('carga error.log', linea) except Exception as ex: print('[Procesando *Error.log] Exception: ', ex) try: info_error = f'IP:[{ip}] - FECHA:[{fecha}] - ' \ f'METODO:[{metodo}] - URL:[{url}]' log_usage('Exception error.log', info_error) except Exception: pass try: with Progress() as prog, session: task1 = prog.add_task( "[yellow bold]Guardando[/yellow bold]", total=len(session.new) ) session.commit() while not prog.finished: prog.update(task1, advance=0.1) time.sleep(0.05) except Exception as ex: log_usage('Exception error.log - Progress session commit', ex) console.print( '[magenta] - Carga completa.. borrando log[/magenta]\n' ) os.remove(log) return True except Exception: console.print( f'[red]Error al intentar abrir/cargar: [{log}[/red]]\n' ) log_usage('Error al abrir/cargar', log) return False else: console.print( f'[bold red]Log: [[/bold red]{log}[bold red]]' ' inexistente.[/bold red]\n' ) log_usage('Log inexistente', log) return False def carga_logs(): """Procesa logs existentes en directorio 'logs_dir', según nombre.""" logpath = logs_dir+'/access.log' if os.path.exists(logpath): carga_access_log(logpath) logpath = logs_dir+'/reverse-access.log' if os.path.exists(logpath): carga_access_log(logpath) logpath = logs_dir+'/error.log' if os.path.exists(logpath): carga_error_logs(logpath) logpath = logs_dir+'/reverse-error.log' if os.path.exists(logpath): carga_error_logs(logpath) def carga_registro_ip(ip_info): """Guarda datos del diccionario ip_info en tabla 'Registro', Actualiza columna 'registro' a '1' en la tabla 'Visita' para IPs guardadas en 'Registro' en esta sessión. """ if not ip_registrada(ip_info['ip']): info_dic = {} info_dic['ip'] = ip_info['ip'] info_dic['hostname'] = ip_info['hostname'] if 'hostname' in ip_info else None info_dic['anycast'] = ip_info['anycast'] if 'anycast' in ip_info else None info_dic['ciudad'] = ip_info['city'] if 'city' in ip_info else None info_dic['region'] = ip_info['region'] if 'region' in ip_info else None info_dic['pais'] = ip_info['country'] if 'country' in ip_info else None info_dic['geoloc'] = ip_info['loc'] if 'loc' in ip_info else None info_dic['organizacion'] = ip_info['org'] if 'org' in ip_info else None info_dic['tzone'] = ip_info['timezone'] if 'timezone' in ip_info else None info_dic['cod_post'] = ip_info['postal'] if 'postal' in ip_info else None try: session.add( Registro( ip=info_dic['ip'], hostname=info_dic['hostname'], anycast=info_dic['anycast'], cuidad=info_dic['ciudad'], region=info_dic['region'], pais=info_dic['pais'], geoloc=info_dic['geoloc'], organizacion=info_dic['organizacion'], fecha_reg=int(time.mktime(time.localtime())), tzone=info_dic['tzone'], cod_post=info_dic['cod_post'], ) ) session.commit() except Exception as ex: print('[session.commit(ADD REGISTRO)] Exception: ', ex) print('Datos-Dic: ', info_dic) stmt = update(Visita).where( Visita.ip == ip_info['ip'] ).values( registro=1 ).execution_options(synchronize_session="fetch") try: session.execute(stmt) session.commit() except Exception as ex: print('[session.commit(UPDT VISITA)] Exception: ', ex) def consulta_ip(ip_consulta, tkn=True): """Consulta API o base de datos por la IPs pasada como argumento, filtra IPs validas antes de proceder. """ if valid_ip(ip_consulta): match tkn: case True: consulta = f'https://ipinfo.io/{ip_consulta}{token}' info_ip = requests.get(consulta).text return loads(info_ip) case False: consulta = f'https://ipinfo.io/{ip_consulta}' info_ip = requests.get(consulta).text return loads(info_ip) case None: resp = consulta_db(ip_consulta) return resp def consulta_db(ip): """Consulta base de datos por la IPs pasada como argumento. Entrega la información de Registro, seguida por todas las visitas y sus detalles. """ try: statement = session.query( Registro, Visita ).join('visitas').filter_by(ip=ip) result = session.execute(statement).all() return result except Exception as ex: print('Exception consulta_db:\n', ex) def registro_ips(): """Consulta API, obtiene datos de IPs en tabla 'Visita' cuya valor en columna 'registro' sea '0'. Utiliza clase Progress() para mostrar el progreso de la transacción. """ statement = select(Visita).filter_by(registro=0) with Progress() as progress: total = len(session.execute(statement).scalars().all()) task1 = progress.add_task( "[bold blue]Cargando [/bold blue]", total=total ) total_ant = total while not progress.finished: res = session.execute(statement).scalars().first() total_act = len(session.execute(statement).scalars().all()) avance = total_ant - total_act total_ant = total_act if res is None: progress.update(task1, advance=avance) else: ip_actual = res.ip ip_info = consulta_ip(ip_actual, True) carga_registro_ip(ip_info) progress.update(task1, advance=avance) console.print( '\n[bold yellow]Registro en base de datos finalizado.[/bold yellow]' ) def mapsgen(): """Realiza 2 consultas de los datos de columna 'geoloc' de la tabla 'Registro' según valor de columna 'cod_html' de la tabla 'Visita', para valores 200 y otros. Llama a función maps_gen con estas listas de valores como argumentos.""" try: loc_200 = get_geoloc(200) loc_300 = get_geoloc(300) maps_gen(loc_200, loc_300) except Exception as ex: print('Exception mapsgen: ', ex)