adelgado/nc_password_client
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

allow same name and add try for other requests

Browse source
This commit is contained in:
Antonio J. Delgado 2024-11-11 10:19:56 +02:00
parent d120396957
commit 88b4dbf276
1 changed files with 150 additions and 115 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

265
nc_password_client/nc_password_client.py
View file

@ -204,127 +204,158 @@ class NextcloudHandler:
def get(self, path): def get(self, path):
'''Do a GET request''' '''Do a GET request'''
r = requests.get( try:
f'{self.http}://{self.host}/{path}', r = requests.get(
auth=(self.user, self.token), verify=self.ssl, headers=self.headers(), f'{self.http}://{self.host}/{path}',
timeout=self.timeout auth=(self.user, self.token), verify=self.ssl, headers=self.headers(),
) timeout=self.timeout
)
if r.status_code == 200: if r.status_code == 200:
return r return r
elif r.status_code == 404: elif r.status_code == 404:
self._log.error(f'File {path} does not exist') self._log.error(
sys.exit(3) 'Path %s does not exist',
else: path
self._log.error(r.status_code) )
sys.exit(4) sys.exit(3)
else:
self._log.error(r.status_code)
sys.exit(4)
except requests.exceptions.ReadTimeout as error:
self._log.error(
"Timeout (%s sec) error doing GET request. %s",
self.timeout,
error
)
def propfind(self, path): def propfind(self, path):
'''Do a PROPFIND request''' '''Do a PROPFIND request'''
s = requests.Session() s = requests.Session()
s.auth = (self.user, self.token) s.auth = (self.user, self.token)
r = s.request( try:
method='PROPFIND', r = s.request(
url=f'{self.http}://{self.host}/{path}', method='PROPFIND',
headers={'Depth': '0'}, url=f'{self.http}://{self.host}/{path}',
data=DIOS_MIO, headers={'Depth': '0'},
verify=self.ssl data=DIOS_MIO,
) verify=self.ssl
)
if r.status_code == 207: if r.status_code == 207:
dom = minidom.parseString(r.text.encode('ascii', 'xmlcharrefreplace')) dom = minidom.parseString(r.text.encode('ascii', 'xmlcharrefreplace'))
try: try:
return { return {
'last_modified': dom.getElementsByTagName( 'last_modified': dom.getElementsByTagName(
'd:getlastmodified' 'd:getlastmodified'
)[0].firstChild.data, )[0].firstChild.data,
'content_type': dom.getElementsByTagName( 'content_type': dom.getElementsByTagName(
'd:getcontenttype' 'd:getcontenttype'
)[0].firstChild.data, )[0].firstChild.data,
'file_id': int(dom.getElementsByTagName( 'file_id': int(dom.getElementsByTagName(
'oc:fileid' 'oc:fileid'
)[0].firstChild.data), )[0].firstChild.data),
'size': int(dom.getElementsByTagName( 'size': int(dom.getElementsByTagName(
'oc:size' 'oc:size'
)[0].firstChild.data), )[0].firstChild.data),
'favorite': int(dom.getElementsByTagName( 'favorite': int(dom.getElementsByTagName(
'oc:favorite' 'oc:favorite'
)[0].firstChild.data), )[0].firstChild.data),
'owner': dom.getElementsByTagName( 'owner': dom.getElementsByTagName(
'oc:owner-display-name' 'oc:owner-display-name'
)[0].firstChild.data, )[0].firstChild.data,
'href': dom.getElementsByTagName( 'href': dom.getElementsByTagName(
'd:href' 'd:href'
)[0].firstChild.data )[0].firstChild.data
} }
except Exception: except Exception:
# I guess it's folder, because it has no content_type # I guess it's folder, because it has no content_type
return { return {
'last_modified': dom.getElementsByTagName( 'last_modified': dom.getElementsByTagName(
'd:getlastmodified' 'd:getlastmodified'
)[0].firstChild.data, )[0].firstChild.data,
'content_type': 'inode/directory', 'content_type': 'inode/directory',
'file_id': dom.getElementsByTagName( 'file_id': dom.getElementsByTagName(
'oc:fileid' 'oc:fileid'
)[0].firstChild.data, )[0].firstChild.data,
'size': dom.getElementsByTagName( 'size': dom.getElementsByTagName(
'oc:size' 'oc:size'
)[0].firstChild.data, )[0].firstChild.data,
'favorite': dom.getElementsByTagName( 'favorite': dom.getElementsByTagName(
'oc:favorite' 'oc:favorite'
)[0].firstChild.data, )[0].firstChild.data,
'owner': dom.getElementsByTagName( 'owner': dom.getElementsByTagName(
'oc:owner-display-name' 'oc:owner-display-name'
)[0].firstChild.data, )[0].firstChild.data,
'href': dom.getElementsByTagName( 'href': dom.getElementsByTagName(
'd:href' 'd:href'
)[0].firstChild.data )[0].firstChild.data
} }
elif r.status_code == 404: elif r.status_code == 404:
return {} return {}
else: else:
self._log.error(
"Nextcloud instance returned status code: %s",
r.status_code
)
except requests.exceptions.ReadTimeout as error:
self._log.error( self._log.error(
"Nextcloud instance returned status code: %s", "Timeout (%s sec) error doing GET request. %s",
r.status_code self.timeout,
error
) )
def put(self, path, src=None): def put(self, path, src=None):
'''Do a PUT request''' '''Do a PUT request'''
if src: try:
r = requests.put( if src:
f'{self.http}://{self.host}/{path}', r = requests.put(
data=open(src, 'rb'), auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout f'{self.http}://{self.host}/{path}',
) data=open(src, 'rb'), auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout
else: )
r = requests.put( else:
f'{self.http}://{self.host}/{path}', r = requests.put(
headers=self.headers, auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout f'{self.http}://{self.host}/{path}',
) headers=self.headers, auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout
)
if r.status_code in [200, 201, 204]: if r.status_code in [200, 201, 204]:
return r, True return r, True
else: else:
self._log.error(
"Nextcloud instance returned status code: %s",
r.status_code
)
except requests.exceptions.ReadTimeout as error:
self._log.error( self._log.error(
"Nextcloud instance returned status code: %s", "Timeout (%s sec) error doing GET request. %s",
r.status_code self.timeout,
error
) )
def delete(self, path): def delete(self, path):
'''Do a DELETE request''' '''Do a DELETE request'''
r = requests.delete( try:
f'{self.http}://{self.host}/{path}', r = requests.delete(
auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout f'{self.http}://{self.host}/{path}',
) auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout
)
if r.status_code in [200, 204]: if r.status_code in [200, 204]:
return r, True return r, True
elif r.status_code == 404: elif r.status_code == 404:
return r, False return r, False
else: else:
self._log.error(
"Nextcloud instance returned status code: %s",
r.status_code
)
except requests.exceptions.ReadTimeout as error:
self._log.error( self._log.error(
"Nextcloud instance returned status code: %s", "Timeout (%s sec) error doing GET request. %s",
r.status_code self.timeout,
error
) )
def talk(self, message, channel): def talk(self, message, channel):
@ -335,21 +366,27 @@ class NextcloudHandler:
} }
spreed_v1_path = "ocs/v2.php/apps/spreed/api/v1/chat" spreed_v1_path = "ocs/v2.php/apps/spreed/api/v1/chat"
try:
r = requests.post(
f'{self.http}://{self.host}/{spreed_v1_path}/{channel}',
data=body,
headers=self.headers(),
auth=(self.user, self.token),
verify=self.ssl, timeout=self.timeout
)
r = requests.post( if r.status_code == 201:
f'{self.http}://{self.host}/{spreed_v1_path}/{channel}', return r, True
data=body, else:
headers=self.headers(), self._log.error(
auth=(self.user, self.token), "Nextcloud instance returned status code: %s",
verify=self.ssl, timeout=self.timeout r.status_code
) )
except requests.exceptions.ReadTimeout as error:
if r.status_code == 201:
return r, True
else:
self._log.error( self._log.error(
"Nextcloud instance returned status code: %s", "Timeout (%s sec) error doing GET request. %s",
r.status_code self.timeout,
error
) )
def request_passwords_session(self): def request_passwords_session(self):
@ -539,8 +576,6 @@ r.status_code
for password in self.list_passwords(): for password in self.list_passwords():
if self.is_same_password(obj, password): if self.is_same_password(obj, password):
return True return True
if password['label'] == obj['label']:
return True
return False return False
def create_password(self, post_obj): def create_password(self, post_obj):