#
# @copyright Copyright (c) 2025, Pietro Marini (pmarini@rcasys.com)
#
# @license GNU AGPL version 3 or any later version
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import random
import string
import sys
import mariadb
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen.canvas import Canvas
import logging
import subprocess
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
params = { "nc_usr": "<nc_usr>",
"nc_pwd": "<nc_pwd>",
"db_user": "<nextcloud_db_usr>",
"db_pwd": "<nextcloud_db_pwd>"
}
# create a series of string, then choose one that will be used to test the indexing
len_single_string = 12
random_strings = ""
nr_strings = 5
for i in range(nr_strings):
random_string = ''.join(random.choices(string.ascii_letters + string.digits, k=len_single_string))
random_strings += " "
random_strings += random_string
test_string = random_strings.split()[random.randint(0,nr_strings-1)]
logging.info("chosen string: {test_string}".format(test_string=test_string))
# create a pdf, testfile.pdf, that contains this series of random strings
my_pdf = Canvas("/tmp/testfile.pdf", pagesize=A4)
my_pdf.drawString(10,10,random_strings)
my_pdf.save()
# look the index for test_string, it should return no documents
########## IMPLEMENT ############
# upload the file testfile.pdf
nc_url = 'http://{nc_usr}:{nc_pwd}@localhost:80/remote.php/dav/files/admin/fts/testfile.pdf'.format(nc_usr=params["nc_usr"],nc_pwd=params["nc_pwd"])
upload_file = subprocess.check_output([
'curl',
'--location',
'--request',
'PUT',
nc_url,
'--header',
'OCS-APIRequest: true',
'--header',
'Destination: {nc_url}'.format(nc_url=nc_url),
'--form',
'file=@"/tmp/testfile.pdf"'
]
, stderr=subprocess.STDOUT)
# search test_string: only the, document testfile.pdf should be returned
search_file = subprocess.check_output([
"occ",
"fulltextsearch:search",
"admin",
test_string
],
stderr=subprocess.STDOUT)
# compare to the file id as found in oc_filecache
try:
conn = mariadb.connect(
user=params["nextcloud_db_usr"],
password=params["nextcloud_db_pwd"],
host=params["nextcloud_db_port"],
port=params["nextcloud_db_port"],
database=params["nextcloud_db"]
)
except mariadb.Error as e:
print(f"Error connecting to database: {e}")
sys.exit(1)
cur = conn.cursor()
cur.execute(
'''
select fileid from oc_filecache
where path = "files/fts/testfile.pdf"
and storage = (select numeric_id from oc_storages where id = "home::admin")'
'''
)
# report the result
# clean-up, delete testfile.pdf locally and on Nextcloud