-3

I'm getting this error message when using yield.

When I remove the yield results and yield timeout the code works fine without the error message

I don't know what is directory or file 'o' since I'm not using it in any way in the code.

here is my full code:

import gradio as gr
import ipaddress
import requests
from requests.auth import HTTPBasicAuth
import os
import string
from datetime import date, datetime

####SETTING UP DATE AND TIME WITH ISRAELI FORMAT###
current_date = date.today()
current_month = current_date.strftime('%B')
current_year = current_date.strftime('%Y')
date_reformat = current_date.strftime('%d/%m/%y')
current_day = current_date.strftime('%d')

###SWITCH###
def switch_ver(ip):
    with open('switches_successful_results.txt','w') as switches_successful, open('switches_failed_results.txt', 'w') as switches_failed:
        ip_addr = ip.split()
        for i in ip_addr:
            ip_addr = list(ipaddress.ip_network(i))
            try:
                basic=HTTPBasicAuth('some','password')
                login = requests.post('http://'+i+':80/rest/v7/login-sessions', auth=basic)
                cookie = login.cookies
                get_ver = requests.get('http://'+i+':80/rest/v7/system/status', cookies=cookie)
                get_ver = get_ver.json()
                get_ver = get_ver['firmware_version']
                get_ver = get_ver
                with open('switches_successful_results.txt', 'a+') as sw:
                    results = 'Switch version for {} is: {} \n'.format(i, get_ver)
                    sw.write(results)
                yield results
                                            
            except requests.exceptions.ConnectTimeout:
                timeout = 'Could not connect to switch: '+i+' REQUEST TIMED OUT\n'
                with open('switches_failed_results.txt', 'a+') as sw:
                    sw.write(timeout)
                yield timeout

    with open('switches_successful_results.txt','r') as switches_successful, open('switches_failed_results.txt', 'r') as switches_failed:
        summary = switches_failed.read() + switches_successful.read()


    return (summary),['switches_successful_results.txt', 'switches_failed_results.txt']
           
###IPBlockerK###
def block_ip(ip):
    duplicate_ips = []
    blocked_ips = []
    invalid_ips = []
    with open('fortigate_ips.txt','r+') as f, open('fortigate_urls.txt', 'r+') as u:
        fortigate_ips = f.read()
        fortigate_urls = u.read()
        ip_addr = ip.split()
        for i in ip_addr:
            try:
                list(ipaddress.ip_network(i))
                if i in fortigate_ips:
                    duplicate_ips.append(i)
                elif ipaddress.ip_address(i).is_private:
                    invalid_ips.append(i)
                else:
                    blocked_ips.append(i)
                    f.write(i + '\n')
            except ValueError:
                if i in fortigate_ips or i in fortigate_urls:
                    duplicate_ips.append(i)
                elif i[0] in string.ascii_letters or i[0] == '*':
                    blocked_ips.append(i)
                    u.write(i + '\n')
                    
                else:
                    invalid_ips.append(i)

    current_time = datetime.now()
    current_time = current_time.strftime('%H:%M:%S')
    if os.path.exists(current_year) == False:
        os.makedirs(current_year + '\\'+ current_month + '\\' + current_day)
        os.chdir(current_year+ '\\' + current_month +'\\'+ current_day)
        with open('Blocked_IPs.txt', 'a+') as Blocked_IPs:
            to_file = ('###############{}###############\n'.format(current_time)+'\n'.join(blocked_ips))+'\n'
            Blocked_IPs.write(to_file)
        os.chdir('D:\\programs\\Python310\\Projects\\net_sec')
    elif os.path.exists(current_year) == True and os.path.exists(current_year + '\\'+ current_month) == False:
        os.chdir(current_year)
        os.makedirs(current_month + '\\' + current_day)
        os.chdir(current_month +'\\'+ current_day)
        with open('Blocked_IPs.txt', 'a+') as Blocked_IPs:
            to_file = ('###############{}###############\n'.format(current_time)+'\n'.join(blocked_ips))+'\n'
            Blocked_IPs.write(to_file)
        os.chdir('D:\\programs\\Python310\\Projects\\net_sec')
    elif os.path.exists(current_year) == True and os.path.exists(current_year + '\\'+ current_month) == True and os.path.exists(current_year + '\\'+ current_month + '\\' + current_day) == False:
        os.chdir(current_year + '\\'+ current_month)
        os.mkdir(current_day)
        os.chdir(current_day)
        with open('Blocked_IPs.txt', 'a+') as Blocked_IPs:
            to_file = ('###############{}###############\n'.format(current_time)+'\n'.join(blocked_ips))+'\n'
            Blocked_IPs.write(to_file)
        os.chdir('D:\\programs\\Python310\\Projects\\net_sec')
    else:
        os.chdir(current_year + '\\' + current_month + '\\' + current_day)
        with open('Blocked_IPs.txt', 'a+') as Blocked_IPs:
            to_file = ('###############{}###############\n'.format(current_time)+'\n'.join(blocked_ips))+'\n'
            Blocked_IPs.write(to_file)
        os.chdir('D:\\programs\\Python310\\Projects\\net_sec')
    
    blocked_ips_result = 'Following IP\s or URLs were Blocked!: \n'+'\n'.join(blocked_ips) +'\n'
    duplicate_ips_result = 'Skipped!...Found duplicates IP\s for: \n'+'\n'.join(duplicate_ips) +'\n'
    invalid_ips_result = 'Skipped!..Invalid IP\s for \n'+'\n'.join(invalid_ips) +'\n'

    with open('fortigate_ips.txt', 'r') as f, open('fortigate_urls.txt', 'r') as u:
        current_commit_stats = len(blocked_ips)
        ips_stats = len(f.readlines())
        urls_stats = len(u.readlines())
        total_stats = ips_stats + urls_stats
               

    if bool(duplicate_ips) == True and bool(blocked_ips) == False:
        print(1)
        return duplicate_ips_result, current_commit_stats, ips_stats, urls_stats, total_stats
    elif bool(duplicate_ips) == True and bool(blocked_ips) == True and bool(invalid_ips) == True:
        print(2)
        return invalid_ips_result + duplicate_ips_result + blocked_ips_result, current_commit_stats, ips_stats, urls_stats, total_stats
    elif bool(invalid_ips) == True and bool(blocked_ips) == True:
        print(3)
        return invalid_ips_result + blocked_ips_result, current_commit_stats, ips_stats, urls_stats, total_stats
    elif bool(invalid_ips) == True and bool(blocked_ips) == True:
        print(4)
        return invalid_ips_result + blocked_ips_result, current_commit_stats, ips_stats, urls_stats, total_stats
    else:
        print(5)
        return (blocked_ips_result), current_commit_stats, ips_stats, urls_stats, total_stats


###GRADIO GUI###
#f = open('fortigate_ips.txt', 'r')
#fortigate = (f.read().split())
#f.close()
with gr.Blocks(title = 'Switcher') as switches_ver:
    gr.Markdown('Welcome to IPBlocker')
    with gr.Tab(label = 'IPBlocker'):
        with gr.Row():
            with gr.Column():
                ips_to_block = gr.Textbox(label = "IPs", lines = 10, placeholder=('Please fill Ips to block'))
                block_btn = gr.Button('Block')
                #ip_lookup = gr.Dropdown(fortigate)
            with gr.Column():
                output_textbox = gr.Textbox(label = "Results", lines=10)
                with gr.Row():
                    current_commit_stats = gr.Textbox(label = 'Current IP\s or URLs added to block:')
                    forti_ips_stats = gr.Textbox(label = 'Total blocked IP\s on Fortigate: ')
                    forti_urls_stats = gr.Textbox(label = 'Total URLs blocked on Fortigate')
                    forti_total_stats = gr.Textbox(label = 'Total blocked IP\s and URLs on Fortigate')
                    block_btn.click(fn=block_ip, inputs = ips_to_block, outputs = [output_textbox, current_commit_stats, forti_ips_stats, forti_urls_stats, forti_total_stats])

    with gr.Tab(label = 'Switcher'):
        with gr.Row():   
            with gr.Column():
                switch_box = gr.Textbox(label = 'Switches', lines = 10, placeholder='Please fill switches IPs...')
                show_ver = gr.Button('Show current switches version')
                upgrade_ver = gr.Button('Upgrade selected switches')
            with gr.Column():
                output_textbox = gr.Textbox(label='Results',lines = 10)
                output_file = gr.File(['switches_successful_results.txt', 'switches_failed_results.txt']) 
                show_ver.click(fn=switch_ver, inputs = switch_box, outputs = [output_textbox, output_file])
                upgrade_ver.click(fn=block_ip, inputs = ips_to_block, outputs=[output_textbox, output_file])
            
            

switches_ver.queue(concurrency_count=20, max_size=20).launch()

full error traceback:

Traceback (most recent call last):
  File "D:\programs\Python310\lib\site-packages\gradio\routes.py", line 273, in run_predict
    output = await app.blocks.process_api(
  File "D:\programs\Python310\lib\site-packages\gradio\blocks.py", line 757, in process_api
    predictions = self.postprocess_data(fn_index, result["prediction"], state)
  File "D:\programs\Python310\lib\site-packages\gradio\blocks.py", line 721, in postprocess_data
    block.postprocess(prediction_value)
  File "D:\programs\Python310\lib\site-packages\gradio\components.py", line 2147, in postprocess
    "name": processing_utils.create_tmp_copy_of_file(
  File "D:\programs\Python310\lib\site-packages\gradio\processing_utils.py", line 323, in create_tmp_copy_of_file
    shutil.copy2(file_path, file_obj.name)
  File "D:\programs\Python310\lib\shutil.py", line 434, in copy2
    copyfile(src, dst, follow_symlinks=follow_symlinks)
  File "D:\programs\Python310\lib\shutil.py", line 254, in copyfile
    with open(src, 'rb') as fsrc:
FileNotFoundError: [Errno 2] No such file or directory: 'o'
LittleWing
  • 21
  • 6

2 Answers2

0

If you are using yield you can iterate only once. It doesn't keep data on memory for all time. Check this out: https://stackoverflow.com/a/231855/17318894

0

The 'o' came from the timeout text "Could not connect..."

From what I understand about gradio, the result, for both yield and return seems to be processed to outputs, which is output_textbox and output_file

As the yield result is timeout (similar goes for results yield case):

  • output_textbox = timeout[0] = 'C'
  • output_file = timeout[1] = 'o'

If you want to remove the errors, you should change the yield result to be compatible to the outputs.

For example:

yield timeout, ['switches_successful_results.txt', 'switches_failed_results.txt']
NoThlnG
  • 297
  • 1
  • 7