Убери лишнее, оптимизируй процесси, оформи как класс, чтоб им можно было пользоваться from elasticsearch import Elasticsearch
import json
import threading
import datetime

es = Elasticsearch(
    ['http://10.0.9.39:9200'],
    use_ssl=False,
)

index_name = 'winlogbeat-2023.12.05'

query_body = {
  "query": {
        "bool": {
            "must": [
                {"term": {"SourceHostname_User": "director"}},
                {
                    "range": {
                        "@timestamp": {
                            "gte": "now-50h"
                        }
                    }
                }
            ]
        }
    },
    "sort": [
        {"@timestamp": "asc"}
    ]
}

page = es.search(
    index=index_name,
    scroll='5m', 
    size=1,    
    body=query_body
)

def scroll(es_obj, _scroll_id, scroll_time='5m'): 
    while True:
        res = es_obj.scroll(scroll_id=_scroll_id, scroll=scroll_time)
        _scroll_id = res['_scroll_id']
        hits = res['hits']['hits']
        if len(hits) == 0: 
            break
        yield from hits

keep_running = True

def check_input():
    global keep_running
    input('Press Enter to stop…\n')
    keep_running = False

threading.Thread(target=check_input).start()

timestamp_cutoff = datetime.datetime.combine(datetime.date.today(), datetime.time(23, 59, 59))

with open('LogsDIR.json', 'w') as f:
    for hit in scroll(es, page['_scroll_id']):
        f.write(json.dumps(hit))
        f.write('\n')
        if not keep_running or datetime.datetime.utcnow() >= timestamp_cutoff:
            break


es.clear_scroll(scroll_id=page['_scroll_id'])
def classify_image(image, user):
    executable_extensions = ['.exe', '.dll', '.sys', '.bat', '.cmd', '.ps1', '.vbs']
    if 'System32' in image or 'AppData' in image:
        if any(image.lower().endswith(ext) for ext in executable_extensions):
            return 3
        else:
            return 2
    elif any(image.lower().endswith(ext) for ext in executable_extensions):
        return 1
    else:
        return 0

def process_record(data, output_file):
    source = data.get('_source', {})
    event_data = source.get('event_data', {})
    user = event_data.get('User', '')
    sourcehostname = event_data.get('SourceHostname', '')
    event_id = source.get('event_id', '')
    thread_id = source.get('thread_id', '')
    image = event_data.get('Image', '')
    utc_time = event_data.get('UtcTime', '')
    classification = classify_image(image, user)
    src_or_user = sourcehostname if sourcehostname else user

    record = {
        "computer_name": src_or_user,
        "EventId": event_id,
        "ThreadId": thread_id,
        "Image":  classification,
        "UtcTime": utc_time,
#        "Class": 
    }
    
    output_file.write(orjson.dumps(record) + b'\n')
    
    return record

input_file_name = 'LogsN.json' 
output_file_name = 'processed_records.jsonl'  

processed_count = 0

with open(input_file_name, 'rb') as input_file, open(output_file_name, 'wb') as output_file:
    for line in input_file:
        try:
            data = orjson.loads(line)
            process_record(data, output_file)
            processed_count += 1
        except KeyError as e:
            print(f"KeyError encountered: {e}")
        
        if processed_count % 10000 == 0:
            print(f'Processed {processed_count} records…')

print(f'Total processed records: {processed_count}')
####################################################################################################################################################################################################################################
input_file_name = 'Processed_records.jsonl'
output_file_name = 'sorted_normalized_records.jsonl'

records = []

with open(input_file_name, 'r', encoding='utf-8') as input_file:
    for line in input_file:
        try:
            record = json.loads(line)
            if all(key in record for key in ['EventId', 'ThreadId', 'Image', 'computer_name', 'UtcTime']) and record['computer_name'] and record['UtcTime'] and str(record['EventId']).isdigit() and str(record['ThreadId']).isdigit() and str(record['Image']).isdigit():  # Проверяем, являются ли значения цифрами

               record['EventId'] = float(record['EventId'])
               record['ThreadId'] = float(record['ThreadId'])
               record['Image'] = float(record['Image'])

               records.append(record)
        except json.JSONDecodeError:
            continue

df = pd.DataFrame(records)

df.sort_values(by=['computer_name', 'UtcTime'], inplace=True)

scaler = MinMaxScaler()
numerical_cols = ['EventId', 'ThreadId', 'Image']
df[numerical_cols] = scaler.fit_transform(df[numerical_cols])

df['Class'] = df.apply(lambda row: 1 if row['computer_name'] in ['director\TestoedovNA', 'director'] else 0, axis=1)

with open(output_file_name, 'w') as output_file:
    for record in df.to_dict(orient='records'):
        json_record = json.dumps(record)
        output_file.write(f"{json_record}\n")

print(f"Total processed and saved records: {len(df)}")