I have a mobile application on kivy, it works well until I try to implement asynchronous execution of site parsing in it, so that the main gui kivy window does not hang and I can interact with it during the execution of a long process. Then the second file (ckecker), site parsing, does not work correctly, especially where the wait for js execution is performed. It either does not wait for js to execute, or waits but with an error somewhere in the process of waiting and outputting the result, as a result the result is incorrect.
I attached a code example without asynchronous execution. I repeat, the second script (checker) works normally only in this mode (when a function is called in it via asyncio.run(), and inside it new_loop) otherwise it does not work normally with the site on which the js script is executed. (Selenium is not recommended).
I also processed my question in chatgpt - it didn't help. I tried the following options: - standard kivymd capabilities (from kivymd.utils.asynckivy import start, sleep) - asyncio (and the entire main.run() in it and purely the load function or only the second script in the call) - threading / concurrent.futures (purely through them and together with asyncio) - asynckivy (which is installed separately) also didn't work Nothing helped. The application managed to start working asynchronously, but the result from the second script was then executed/came incorrectly.
main.py
from kivy.lang import Builder
from kivy.core.clipboard import Clipboard as Cb
from kivy.core.window import Window
from kivy.properties import ObjectProperty
from kivy.utils import platform
from kivy.clock import Clock
from kivy.uix.modalview import ModalView
from kivy.uix.label import Label
from kivymd.app import MDApp
from kivymd.uix.menu import MDDropdownMenu
from kivymd.uix.button import MDRoundFlatIconButton
from kivymd.uix.scrollview import MDScrollView
import cheker as checkN
KV = '''
<ContentNavigationDrawer>
MDList:
OneLineIconListItem:
text: "Home"
theme_text_color: "Custom"
text_color: 1, 1, 1, 1
on_press:
root.nav_drawer.set_state("close")
root.screen_manager.current = "scr 1"
IconLeftWidget:
icon: 'home'
MDScreen:
MDBoxLayout:
id: home
orientation: 'vertical'
MDTopAppBar:
pos_hint: {"top": 1}
elevation: 4
title: "OSINT"
left_action_items: [["menu", lambda x: nav_drawer.set_state("open")]]
MDNavigationLayout:
id: navig
MDScreenManager:
id: screen_manager
MDScreen:
id: scr_1
name: "scr 1"
MDBottomNavigation:
id: sc_0
selected_color_background: "orange"
text_color_active: "lightgrey"
MDBottomNavigationItem:
id: sc_1
name: 'screen 1'
text: 'Home'
icon: 'home'
MDRoundFlatIconButton:
id: lst1
text: 'Choose option'
pos_hint: {'center_x':0.5, 'center_y':0.7}
on_release: app.menu.open()
MDTextField:
id: txt
hint_text: ''
helper_text: ''
helper_text_mode: 'persistent'
text: ''
mode: 'rectangle'
pos_hint: {'center_x':0.5, 'center_y':0.45}
size_hint: 0.28, 0.15
MDRoundFlatIconButton:
id: startbutton
text: 'Start'
pos_hint: {'center_x':0.5, 'center_y':0.20}
on_release: app.start()
MDBottomNavigationItem:
id: sc_3
name: 'screen 3'
text: 'Result'
icon: 'text'
ScrollView:
MDLabel:
id: scroll
markup: True
text: ''
theme_text_color: "Custom"
text_color: 1, 1, 1, 1
font_size: '15sp'
multiline: True
text_size: self.width, None
size_hint_y: None
height: self.texture_size[1]
on_touch_down: app.linki(app.bufftext)
on_ref_press: app.webopen(args[1])
MDNavigationDrawer:
id: nav_drawer
radius: (0, 16, 16, 0)
ContentNavigationDrawer:
screen_manager: screen_manager
nav_drawer: nav_drawer
'''
class MyApp(MDApp):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.list_info = ['Number']
self.dl = Window.width
self.sh = Window.height
self.screen = Builder.load_string(KV)
menu_items = [
{
'id': 'm1',
'text': f'{i}',
'viewclass': 'OneLineListItem',
'on_release': lambda x=f'{i}': self.menu_callback(x),
} for i in self.list_info
]
self.menu = MDDropdownMenu(
caller=self.screen.ids.lst1,
items=menu_items,
width_mult=5,
)
def menu_callback(self, text_item):
self.screen.ids.lst1.text = text_item
self.screen.ids.txt.hint_text = text_item
lst2 = self.screen.ids.sc_1
self.menu.dismiss()
def build(self):
self.theme_cls.material_style = 'M3'
self.theme_cls.theme_style = 'Dark'
self.title = 'OSINT'
return self.screen
def start(self):
self.screen.ids.sc_0.switch_tab('screen 3')
self.root.ids.scroll.text = f'Please wait'
Clock.schedule_once(lambda dt: self.load(), 1.0)
def load(self):
self.res = ''
opt1 = self.screen.ids.lst1.text
opt3 = self.screen.ids.txt.text
if opt1 != 'Choose option':
if opt1 == 'Number':
self.res = checkN.parse_num(opt3)
else:
self.root.ids.scroll.text = 'Select/enter the correct data.'
if self.res != '':
self.root.ids.scroll.text = self.res
else:
self.root.ids.scroll.text = 'No results'
if __name__ == "__main__":
MyApp().run()
cheker.py
import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
import asyncio
from requests_html import HTMLSession, AsyncHTMLSession
def parse_num(numbers):
try:
res1 = europe(numbers)
except:
res1 = f'{color.RED}[info]:{color.END}\nError internet/proxy_server/site. Try again.\n'
try:
res2 = asyncio.run(all_num(numbers))
except:
res2 = f'{color.RED}[whosenumber.info]:{color.END}\nError internet/proxy_server/site. Try again.\n'
try:
res4 = europe(numbers)
except:
res4 = f'{color.RED}[ua.tellows.org]:{color.END}\nError internet/proxy_server/site. Try again.\n'
res = res1+res2+res4
return res
async def all_num(numbers):
uu='https://whosenumber.info'
headers = {'User-Agent': f'{UserAgent().random}', }
url = f'{uu}/'
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
session = AsyncHTMLSession()
response = session.get(f'{url}{numbers}', headers=headers, timeout=60)
await response.html.arender(sleep=10)
soup = BeautifulSoup(response.html.html, 'html.parser')
try:
divF_tag = soup.find('div', attrs={'style':'padding:10px; padding-left:20px;'})
ttl = ''
try:
divS_tag = divF_tag.find('span', attrs={'id':'t_reiting'})
ttl = divS_tag.get_text()
except:
pass
comm = ''
try:
ij = 1
for p_tag in divF_tag.findAll('div', attrs={'style': 'margin-left:15px;margin-top:10px;color:#6a6a6a; padding-bottom:10px;'}):
try:
pp_tag = p_tag.get_text()
pp_tag = pp_tag.replace('<br/>', '')
pp_tag = pp_tag.replace('<br>', '')
pp_tag = pp_tag.replace('<b/>', '')
pp_tag = pp_tag.replace('<b>', '')
pp_tag = pp_tag.replace('</b>', '')
pp_tag = pp_tag.replace('</br>', '')
comm += f' [{ij}]: {pp_tag}\n'
ij+=1
except:
continue
except:
pass
logger.info('all - there is a result')
resF = (f'{color.RED}[whosenumber.info] - info:{color.END}\n'
f'{color.BOLD}Rating: {color.END}{ttl}\n{color.BOLD}Comments: {color.END}\n{comm}\n')
except:
logger.info('all - result error')
resF = (f'{color.RED}[whosenumber.info] - info:{color.END}\n'
f'Error internet/proxy_server/site. Try again.\n')
res = f'\n{resF}'
await session.close()
new_loop.close()
return res
def europe(numbers):
uu = 'https://ua.tellows.org'
headers = {'User-Agent': f'{UserAgent().random}', }
url = f'{uu}/num/'
response = requests.get(f'{url}{numbers}', headers=headers, timeout=60)
soup = BeautifulSoup(response.text, 'html.parser')
try:
ttl = ""
comm = ""
for divF_tag in soup.findAll('div', attrs={'id': 'userratings'}):
try:
for divS_tag in divF_tag.findAll('div', attrs={'class': 'col-md-4 mt-2'}):
try:
for h5_tag in divS_tag.findAll('h5'):
ttl += f"{color.BOLD}{h5_tag.get_text()}{color.END}\n"
for th_tag in divS_tag.findAll('th'):
tt = str(th_tag).replace("<th>\n ", "")
tt = tt.replace(" ", "")
tt = tt.replace('<br/>\n<span class="small">', ": ")
tt = tt.replace("</span>\n</th>", "")
ttl += f" {tt}\n"
except:
continue
except:
continue
for comms_tag in soup.findAll('ol', attrs={'id': 'singlecomments'}):
try:
ii = 1
comm += f"{color.BOLD}Comments: {color.END}\n"
for comm_tag in comms_tag.findAll('div', attrs={'class': 'col comment-body'}):
try:
for pCom_tag in comm_tag.findAll('p', attrs={'class': 'mb-0'}):
comm += f" [{ii}]{pCom_tag.get_text()}\n"
ii += 1
except:
continue
try:
jj = 1
for dCom_tag in comm_tag.findAll('div', attrs={'class': 'ccomment'}):
try:
comm += f" [0{jj}]{dCom_tag.get_text()}\n"
jj += 1
except:
continue
except:
continue
except:
continue
logger.info('Europe - there is a result')
resF = (f'{color.RED}[ua.tellows.org] - info:{color.END}\n'
f'{ttl}\n{comm}\n')
except:
logger.info('Europe - result error')
resF = (f'{color.RED}[ua.tellows.org] - info:{color.END}\n'
f'{color.BOLD}Info: {color.END}Error internet/proxy_server/site. Try again.\n')
res = f'\n{resF}'
response.close()
return res
Please test the source code I provided and try to run asynchronous execution of kivy, so that the gui window does not freeze, but at the same time the second script is executed normally (the result came from parsing the site), especially in the function where the expectation of the site js is executed. For the test, the number that needs to be entered in the application: +37129502847. I took it from the parsed site, so that I would definitely get the result.
fjl
|
4966
posts
|
PythonAnywhere staff
|
