forked from JanHelbling/py9kw
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathpy9kw.py
executable file
·680 lines (606 loc) · 28.8 KB
/
py9kw.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# py9kw.py - A API for the Captcha-solvingservice 9kw.eu
#
# Copyright (C) 2014 by Jan Helbling <jan.helbling@mailbox.org>
# Updted 2020-01-25 by over_nine_thousand
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import argparse
import json
import logging
import re
import time
from base64 import b64encode
from typing import Union
from urllib.parse import urlencode
from enum import Enum
import httpx
from pydantic import BaseModel, HttpUrl, ValidationError
class CaptchaFeedback(Enum):
CAPTCHA_CORRECT = 1
CAPTCHA_INCORRECT = 2
CAPTCHA_ABORT_CURRENT_CAPTCHA = 3
class CaptchaType(Enum):
STANDARD = 0
ROTATE = 1
TEXT = 2
MOUSE = 3
MULTIMOUSE = 4
PUZZLE = 5
AUDIO = 6
OVERSIZE = 7
INTERACTIVE = 8
class NumericLimit:
NONE = 0
ONLY_NUMBERS = 1
ONLY_CHARACTERS = 2 # A-Za-z
ONLY_CHARACTERS_AND_NUMBERS = 3 # A-Za-z0-9
ONLY_NUMBERS_AND_COMMA_SPACE_MINUS = 4
ONLY_CHARACTERS_AND_NUMBERS_AND_COMMA_SPACE_MINUS = 5
def printInfo(msg):
print('[py9kw] %s' % msg)
# See API docs: https://www.9kw.eu/api.html
API_BASE = 'https://www.9kw.eu/index.cgi'
# Parameter used as 'source' in all API requests
API_SOURCE = 'py9kw-api'
# Values according to website 2020-01-25
PARAM_PRIO_MAX = 20
PARAM_PRIO_DEFAULT = 0
PARAM_MAXTIMEOUT_MIN = 60
PARAM_MAXTIMEOUT_DEFAULT = 600
PARAM_MAXTIMEOUT_MAX = 3999
PARAM_MIN_CREDITS_TO_SOLVE_ONE_CAPTCHA = 10
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.DEBUG)
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.WARNING)
headers = {'User-Agent': f'Python-urllib/3.x ({API_SOURCE})'}
class Py9kw:
def __init__(self, apikey: str):
"""Initialize py9kw with APIKEY """
self.apikey = apikey
self.selfsolve = None
self.prio = None
self.maxtimeout = None
self.numericLimit = NumericLimit.NONE
self.minCaptchaResultCharacterLength = None
self.maxCaptchaResultCharacterLength = None
self.imagedata = None
self.waitSecondsPerLoop = None
self.extrauploaddata = {}
self.sleepOutputFrequencySeconds = None
self.captchaType = CaptchaType.STANDARD
self.oldsource = None
self.pageurl = None
self.userAgent = None
self.captchaid = None
self.credits = -1
# Custom errors also possible besides known API errorcodes e.g. 600 --> "ERROR_NO_USER" --> See README.md
self.errorcode = -1
self.errormsg = None
self.response = {} # Last server json response
self.storedCaptchaResult = None
self.exceptionHandling = False
def callAPI(self, url: str):
timeout = httpx.Timeout(120.0, read=120.0, connect=120.0)
req = httpx.get(url=url, headers=headers, timeout=timeout)
logging.debug('Response: ' + req.text)
return req.json()
def setRecaptchaV2Params(self, siteURL: str, siteKey: str, userAgent: str = None):
self.setInteractiveCaptchaParams(captchaType='recaptchav2', siteURL=siteURL, siteKey=siteKey, userAgent=userAgent)
def setRecaptchaV3Params(self, siteURL: str, siteKey: str, userAgent: str = None):
self.setInteractiveCaptchaParams(captchaType='recaptchav3', siteURL=siteURL, siteKey=siteKey, userAgent=userAgent)
def setHcaptchaParams(self, siteURL: str, siteKey: str, userAgent: str = None):
self.setInteractiveCaptchaParams(captchaType='hcaptcha', siteURL=siteURL, siteKey=siteKey, userAgent=userAgent)
def setInteractiveCaptchaParams(self, captchaType: str, siteKey: str, siteURL: str, userAgent: str = None):
self.setImagedata(imagedata=siteKey)
self.setUseragent(userAgent=userAgent)
self.oldsource = captchaType
self.setPageurl(pageurl=siteURL)
self.captchaType = CaptchaType.INTERACTIVE
def setImageCaptchaParams(self, imagedata: Union[bytes, str]):
""" You can provide an URL to an image or a file. """
self.setImagedata(imagedata)
self.captchaType = CaptchaType.STANDARD
def setImagedata(self, imagedata: Union[bytes, str]):
""" You can provide an URL to an image or a file. """
self.imagedata = imagedata
def checkError(self, response: dict):
""" Checks for errors in json response and returns error_code(int) and error_message(String) separated as API returns them both in one String. """
error_plain = response.get('error', None)
if error_plain is not None:
# Error found
logging.info('[checkError] Found error: Plain error: ' + error_plain)
try:
error_MatchObject = re.compile(r'^(\d{4}) (.+)').search(error_plain)
self.errorcode = int(error_MatchObject.group(1))
self.errormsg = error_MatchObject.group(2)
logging.info('[checkError] Detected error-response: Number: %d | Message: %s' % (self.errorcode, self.errormsg))
except:
# This should never happen
self.errormsg = 'Error while parsing error number and message'
self.errorcode = 666
logging.warning(self.errormsg)
else:
# No error found
# Reset error state
self.errorcode = -1
self.errormsg = None
return self.errorcode, self.errormsg
def setCaptchaType(self, newCaptchaType: CaptchaType):
self.captchaType = newCaptchaType
def getCaptchaCost(self) -> int:
"""Returns how much credits it would cost to solve one captcha with current priority setting."""
captchacost = 10
# This may change at any time. See table of cost on 9kw.eu main page.
if self.captchaType == CaptchaType.AUDIO or self.captchaType == CaptchaType.OVERSIZE:
captchacost = 12
elif self.captchaType == CaptchaType.INTERACTIVE:
captchacost = 25
captchacost += self.getPrio()
return captchacost
def setSelfsolve(self, paramselfsolve: bool):
""" Set this to true to testing:
All captchas sent via given apikey will only be presented to the owner of that apikey for solving. """
self.selfsolve = paramselfsolve
def setPriority(self, prio: int):
self.prio = prio
def setPageurl(self, pageurl: str):
self.pageurl = pageurl
def setUseragent(self, userAgent: str):
""" User-Agent value which 9kw.eu will use to solve captcha.
Sometimes required for solving interactive captchas. """
self.userAgent = userAgent
def getPrio(self) -> int:
if not isinstance(self.prio, int):
return PARAM_PRIO_DEFAULT
if self.prio > PARAM_PRIO_MAX:
# Fallback
return PARAM_PRIO_MAX
elif self.prio > 0:
return self.prio
else:
return PARAM_PRIO_DEFAULT
def canSolveOneMoreCaptcha(self) -> bool:
""" Returns True if there are enough credits available to solve one more captcha. Depending on the currently set priority. """
if self.getcredits() >= self.getCaptchaCost():
return True
else:
return False
def setWaitSecondsPerLoop(self, waitSeconds: int):
self.waitSecondsPerLoop = waitSeconds
def getWaitSecondsPerLoop(self) -> int:
if self.waitSecondsPerLoop is not None and self.waitSecondsPerLoop > 0:
return self.waitSecondsPerLoop
else:
# Fallback / default
return 10
def setAdditionalCaptchaUploadParams(self, uploaddata: dict):
""" Use this to add extra captcha upload parameters such as: 'case-sensitive':1.
"""
self.extrauploaddata = uploaddata
def setTimeout(self, maxtimeout: int):
""" Defines how many seconds the server will allow users to solve the uploaded captcha before giving up. """
self.maxtimeout = maxtimeout
def setNumericLimit(self, numericLimit: int):
""" Limits allowed characters of expected captcha result. """
self.numericLimit = numericLimit
def getNumericLimit(self) -> int:
if not isinstance(self.numericLimit, int):
return NumericLimit.NONE
else:
return self.numericLimit
def setMaxCaptchaResultCharacterLength(self, minLength: int):
self.minCaptchaResultCharacterLength = minLength
def getMaxCaptchaResultCharacterLength(self) -> int:
if not isinstance(self.maxCaptchaResultCharacterLength, int):
return 1
if self.maxCaptchaResultCharacterLength < 1:
return 1
elif self.maxCaptchaResultCharacterLength > 99:
return 99
else:
return self.maxCaptchaResultCharacterLength
def getMinCaptchaResultCharacterLength(self) -> int:
if not isinstance(self.minCaptchaResultCharacterLength, int):
return 1
if self.minCaptchaResultCharacterLength < 1:
return 1
elif self.minCaptchaResultCharacterLength > 20:
return 20
else:
return self.maxCaptchaResultCharacterLength
def setMinCaptchaResultCharacterLength(self, maxLength: int):
self.maxCaptchaResultCharacterLength = maxLength
def setExactExpectedCaptchaResultCharacterLength(self, length: int):
""" Use this if you know the exact expected length of your captcha result. """
self.setExpectedCaptchaResultLengthLimits(minlen=length, maxlen=length)
def setExpectedCaptchaResultLengthLimits(self, minlen: int, maxlen: int):
""" Wrapper to define character limits of captcha result. """
self.setMinCaptchaResultCharacterLength(minlen)
self.setMaxCaptchaResultCharacterLength(maxlen)
def setSleepOutputFrequency(self, outputSeconds: int):
""" Defines output frequency for all loops containing sleep statements. Higher = Less output -> less "comsole-spam". """
self.sleepOutputFrequencySeconds = outputSeconds
def getSleepOutputFrequency(self) -> int:
if not isinstance(self.sleepOutputFrequencySeconds, int):
# Fallback/default
return 10
else:
return self.sleepOutputFrequencySeconds
def setCaptchaID(self, captchaID: int):
""" Use this if you e.g. built an application which saves a captchaID to e.g. resume on crash without wasing credits. """
self.captchaid = captchaID
def getTimeout(self) -> Union[int, None]:
if self.maxtimeout is None:
return PARAM_MAXTIMEOUT_DEFAULT
if self.maxtimeout < PARAM_MAXTIMEOUT_MIN:
return PARAM_MAXTIMEOUT_MIN
elif self.maxtimeout > PARAM_MAXTIMEOUT_MAX:
return PARAM_MAXTIMEOUT_MAX
else:
return self.maxtimeout
def setEnableExceptionHandling(self, enable: bool):
self.exceptionHandling = enable
def solveCaptcha(self) -> Union[str, None]:
self.uploadCaptcha()
return self.obtainResultAuto()
def uploadCaptcha(self, store_image_path=None) -> int:
"""Uploads the pre given Captcha to 9kw.eu.
Returns captchaid
"""
logger_prefix = '[uploadcaptcha]'
if self.imagedata is None:
logging.info(f'{logger_prefix} imagedata is None')
return -1
# Step 1: Set optional parameters and check if user has enough credits
logging.info(logger_prefix + ' Attempting to upload captcha...')
if not self.canSolveOneMoreCaptcha():
# Developer/user mistake
logging.info(logger_prefix + ' Not enough credits to solve captcha')
return -1
# Step 2: Prepare image data we want to upload
# First check if we have an URL --> Download image first
class Link(BaseModel):
url: HttpUrl
thisimagedata = self.imagedata
try:
Link(url=thisimagedata)
urlValid = True
except ValidationError:
urlValid = False
if urlValid:
# Download image so we can upload it to captchasolver
logging.info(logger_prefix + ' Provided source is an URL: ' + thisimagedata)
thisimagedata = downloadCaptchaImageFromWebsite(thisimagedata, store_image_path)
# Step 3: Prepare all parameters we want to send
# First add all necessary fields
getdata = {'action': 'usercaptchaupload', 'apikey': self.apikey, 'source': API_SOURCE, 'json': 1}
if isinstance(thisimagedata, bytes):
# Image file --> Base64
thisimagedata = b64encode(thisimagedata)
getdata['base64'] = 1
else:
# String -> Probably interactive captcha
# Default serverside = 0 -> Do not include field at all
# getdata['base64'] = '0'
pass
getdata['file-upload-01'] = thisimagedata
# Now add all optional fields
if self.maxtimeout is not None:
getdata['maxtimeout'] = self.getTimeout()
if self.prio is not None:
# getdata['prio'] = self.getPrio()
getdata['prio'] = self.getPrio()
if self.captchaType == CaptchaType.INTERACTIVE:
getdata['interactive'] = 1
if self.oldsource is not None:
getdata['oldsource'] = self.oldsource
if self.pageurl is not None:
getdata['pageurl'] = self.pageurl
if self.userAgent is not None:
getdata['useragent'] = self.userAgent
if self.selfsolve is True:
getdata['selfsolve'] = 1
if self.numericLimit != NumericLimit.NONE:
getdata['numeric'] = self.numericLimit
if self.minCaptchaResultCharacterLength is not None:
getdata['min_len'] = self.getMinCaptchaResultCharacterLength()
if self.maxCaptchaResultCharacterLength is not None:
getdata['max_len'] = self.getMaxCaptchaResultCharacterLength()
# Add all additional fields that are given via additional dict
if isinstance(self.extrauploaddata, dict) and len(self.extrauploaddata) > 0:
getdata.update(self.extrauploaddata)
logging.info(logger_prefix + ' extra params:\n' + json.dumps(self.extrauploaddata, indent=4))
# Step 4: Send data and return captchaid
logging.info(f'Sending captchadata to 9kw: {getdata}')
jsonResponse = self.callAPI(url=API_BASE + '?' + urlencode(getdata))
logging.debug(logger_prefix + 'json debug: ' + str(jsonResponse))
response = jsonResponse
self.checkError(response)
self.captchaid = int(response.get('captchaid', -1))
if self.errorcode > -1 or self.captchaid == -1:
logging.warning(logger_prefix + 'Error happened and/or did not get captchaid')
return -1
logging.info(f'{logger_prefix} [DONE] Uploaded => captchaid: {self.captchaid}')
return self.captchaid
def obtainResultAuto(self) -> Union[str, None]:
"""Wait until the Captcha is solved and return result."""
logger_prefix = '[sleepAndGetResult] '
if self.captchaid is None:
# Developer mistake
raise Exception("captchaid cannot be None here")
waitSecondsPerLoop = self.getWaitSecondsPerLoop()
logging.info(logger_prefix + 'Waiting until the Captcha is solved or maxtimeout %d (includes %d extra seconds) has expired ...' % (self.getTimeout(), waitSecondsPerLoop))
total_time_waited = 0
waitSecondsLeft = self.getTimeout()
lastOutputSecondsAgo = self.getSleepOutputFrequency()
while waitSecondsLeft > 0:
if lastOutputSecondsAgo >= self.getSleepOutputFrequency():
logging.info(logger_prefix + 'Waiting for result | Seconds left: %d / %d' % (waitSecondsLeft, self.getTimeout()))
lastOutputSecondsAgo = 0
captchaResult = self.obtainResult()
tryAgainStatus = self.response.get('try_again', False)
if captchaResult is not None:
# We've reached our goal :)
logging.info(logger_prefix + 'Done! Total seconds waited for result: %d' % total_time_waited)
return captchaResult
if self.errorcode > -1 and self.errorcode != 602:
# Retry only on 602 NO_ANSWER_YET - step out of loop if any other error happens
logging.info(logger_prefix + 'Error happened: ' + str(self.errorcode) + ' --> Giving up')
break
elif tryAgainStatus == 0:
logging.info(logger_prefix + 'Server does not want us to try again --> Stopping')
break
if waitSecondsLeft >= waitSecondsPerLoop:
thisSecondsWait = waitSecondsPerLoop
else:
thisSecondsWait = waitSecondsLeft
logging.info(logger_prefix + 'Waiting %d seconds' % thisSecondsWait)
time.sleep(thisSecondsWait)
total_time_waited += thisSecondsWait
waitSecondsLeft -= thisSecondsWait
lastOutputSecondsAgo += thisSecondsWait
logging.info(logger_prefix + 'Time expired! Failed to find result!')
self.errorcode = 601
self.errormsg = 'ERROR_INTERNAL_TIMEOUT'
if self.exceptionHandling:
raise Exception(self.errormsg)
else:
return None
def obtainResult(self) -> Union[str, None]:
"""Get result from 9kw.eu. Use obtainResultAuto for auto-wait handling! """
logger_prefix = '[getresult] '
if self.captchaid is None:
raise Exception("captchaid cannot be None here")
getdata = {
'action': 'usercaptchacorrectdata',
'id': self.captchaid,
'apikey': self.apikey,
'info': 1,
'source': API_SOURCE,
'json': 1
}
logging.info(f'{logger_prefix} Try to fetch the solved result from 9kw.eu API | captchaid: {self.captchaid}')
self.response = self.callAPI(url=API_BASE + '?' + urlencode(getdata))
self.checkError(self.response)
answer = self.response.get('answer', None)
nodata = self.response.get('nodata', -1)
thiscredits = self.response.get('credits', -1)
if thiscredits != -1:
# 2020-02-06: API might sometimes return this as a String although it is supposed to be a number
if isinstance(thiscredits, str):
thiscredits = int(thiscredits)
# Update credits value on change
if thiscredits != self.credits:
logging.info(f'{logger_prefix} Updated credits value from old: {self.credits} to new: {thiscredits}')
self.credits = thiscredits
if nodata == 1:
logging.info(logger_prefix + 'No answer yet')
self.setErrorCode(602)
self.errormsg = 'NO_ANSWER_YET'
# Do not throw exception here because this simply means that we did not get a result yet.
return None
elif answer is not None and answer == 'ERROR NO USER':
# Special: We need to set an error to make sure that our sleep handling would stop!
self.setErrorCode(2700)
self.errormsg = 'ERROR_NO_USER'
logging.info(logger_prefix + 'No users there to solve at this moment --> Or your timeout is too small OR you\'ve aborted this captcha before.')
if self.exceptionHandling:
raise Exception(self.errormsg)
else:
return None
elif self.errorcode > -1:
logging.info(logger_prefix + 'Error %d: %s' % (self.errorcode, self.errormsg))
if self.exceptionHandling:
raise Exception(self.errormsg)
else:
return None
elif answer is None:
# Answer is not given but also we did not get any errormessage
logging.warning(logger_prefix + '[FAILURE] --> Failed to find answer --> Unknown failure')
if self.exceptionHandling:
raise Exception(self.errormsg)
else:
return None
else:
# Captcha-Answer is given
logging.info(logger_prefix + '[SUCCESS] Captcha solved! captchaid %d --> Answer: \'%s\'' % (self.captchaid, answer))
self.storedCaptchaResult = None
return answer
def getStoredResult(self) -> str:
""" Returns result of captcha if it has been solved before. """
return self.storedCaptchaResult
def setCaptchaCorrect(self, iscorrect: bool) -> bool:
"""Send feedback, is the Captcha result correct or not?"""
logger_prefix = '[captcha_correct] '
if iscorrect:
logging.info(logger_prefix + 'Sending POSITIVE captcha solved feedback ...')
return self.sendCaptchaFeedback(CaptchaFeedback.CAPTCHA_CORRECT.value)
else:
logging.info(logger_prefix + 'Sending NEGATIVE captcha solved feedback ...')
return self.sendCaptchaFeedback(CaptchaFeedback.CAPTCHA_INCORRECT.value)
def abortCaptcha(self) -> bool:
"""Send feedback, aborts the already sent captcha. If no answer is available yet, no credits will be used in this case!"""
logging.info('Aborting captcha...')
return self.sendCaptchaFeedback(CaptchaFeedback.CAPTCHA_ABORT_CURRENT_CAPTCHA.value)
def sendCaptchaFeedback(self, captchaFeedbackNumber) -> bool:
"""Send feedback, is the Captcha result correct(=1) or not(=2) or does the user want to abort(=3)?"""
logger_prefix = '[sendCaptchaFeedback] '
logging.info(logger_prefix + 'Sending captcha feedback : %d' % captchaFeedbackNumber)
if self.captchaid is None or self.captchaid <= 0:
# Developer mistake
logging.warning(logger_prefix + 'Cannot send captcha feedback because captchaid is None')
return False
getdata = {
'action': 'usercaptchacorrectback',
'correct': captchaFeedbackNumber,
'id': self.captchaid,
'apikey': self.apikey,
'source': API_SOURCE,
'json': 1
}
try:
thisjson = self.callAPI(API_BASE + '?' + urlencode(getdata))
# Check for errors but do not handle them. If something does wrong here it is not so important!
self.checkError(thisjson)
return True
except:
logging.warning(logger_prefix + 'Error in captcha_correct')
return False
def getcredits(self):
"""Get aviable Credits..."""
logger_info = '[getcredits] '
logging.info(f'{logger_info} Fetching available credits...')
getdata = {
'action': 'usercaptchaguthaben',
'apikey': self.apikey,
'source': API_SOURCE,
'json': 1
}
response = self.callAPI(url=API_BASE + '?' + urlencode(getdata))
self.checkError(response)
if self.errorcode > -1:
logging.warning(logger_info + 'Failed to obtain credits: %s' % self.errormsg)
return -1
usercredits = response.get('credits', -1)
self.credits = usercredits
logging.info(f'{logger_info} {self.credits} credits available | Cost per captcha (with current prio {self.getPrio()}): {self.getCaptchaCost()} | Enough to solve approximately {self.credits / self.getCaptchaCost()} captchas')
return self.credits
def setErrorCode(self, errorcode: int):
self.errorcode = errorcode
def getErrorCode(self) -> int:
return self.errorcode
def resetSolver(self):
""" Call this to reset all runtime values if you e.g. want to re-use a previously created solver instance while keeping your settings (prio, maxtimeout and so on). """
self.captchaid = None
self.errorcode = -1
self.errormsg = None
self.storedCaptchaResult = None
return
def downloadCaptchaImageFromWebsite(image_url: str, image_path: Union[str, None] = None) -> bytes:
""" Downloads captcha image from website and returns the file.
This optionally saves the image to <image_path>. """
try:
req = httpx.get(url=image_url)
imagefile = req.read()
# Save file only if path is given
if image_path is not None:
with open(image_path, 'wb') as file:
file.write(imagefile)
logging.info('[getCaptchaImageFromWebsite] [OK]')
return imagefile
except IOError as ioe:
logging.info('[getCaptchaImageFromWebsite] [FAIL]')
raise ioe
if __name__ == '__main__':
from sys import argv
my_parser = argparse.ArgumentParser()
my_parser.add_argument('-a', '--apikey', help='9kw.eu API Key', type=str, required=True)
my_parser.add_argument('-t', '--timeoutseconds', help='Max time (seconds) to solve', type=int, default=None)
args = my_parser.parse_args()
# Define exactly what we expect as a result according to: https://www.9kw.eu/api.html#apisubmit-tab
selfsolve = True
captchaSolver = Py9kw(args.apikey)
captchaSolver.setNumericLimit(numericLimit=NumericLimit.ONLY_CHARACTERS_AND_NUMBERS)
# Our test captcha image contains exactly seven characters so let's define the expected captcha answer length
captchaSolver.setMinCaptchaResultCharacterLength(7)
captchaSolver.setMaxCaptchaResultCharacterLength(7)
# Alternatively (in this case) we can use this:
captchaSolver.setExactExpectedCaptchaResultCharacterLength(7)
if selfsolve:
# Captcha will only be presented to owner of apikey for solving.
captchaSolver.setSelfsolve(True)
# additionalParams['nomd5'] = 1
# captchaSolver.setAdditionalCaptchaUploadParams({'phrase': 1})
# captchaSolver.setWaitSecondsPerLoop(5)
if args.timeoutseconds is not None:
captchaSolver.setTimeout(args.timeoutseconds)
# captchaSolver.setSleepOutputFrequency(10)
# captchaSolver.setPriority(20)
# Get a Sample-Captcha
creditsLeftBefore = captchaSolver.getcredits()
if not captchaSolver.canSolveOneMoreCaptcha():
print('[py9kw-test] Not enough Credits! < %d' % captchaSolver.getCaptchaCost())
exit(0)
print('[py9kw-test] Credits: {}'.format(creditsLeftBefore))
sampleCaptchaUrl = 'https://confluence.atlassian.com/download/attachments/216957808/captcha.png?version=1&modificationDate=1272411042125&api=v2'
captchaSolver.setImagedata(imagedata=sampleCaptchaUrl)
# Example code to show how to abort captchas.
abortCaptcha = False
if abortCaptcha:
print('Trying to abort already uploaded captcha --> No credits should be used')
captchaAborted = captchaSolver.abortCaptcha()
if captchaAborted:
print('Successfully aborted captcha')
else:
# This should never happen
print('Failed to abort captcha')
exit(0)
# Sleep and get result
result = captchaSolver.solveCaptcha()
# Evaluate Result
if result is None:
printInfo('[py9kw-test] No result --> END')
exit(1)
printInfo('[py9kw-test] String returned!')
expectedAnswer = 'viearer'
printInfo(f'[py9kw-test] Checking if the received string is "{expectedAnswer}"...')
result_is_correct = False
if result.lower() == expectedAnswer:
printInfo('[py9kw-test] Test passed --> executing captcha_correct')
captchaSolver.setCaptchaCorrect(True)
result_is_correct = True
printInfo('[py9kw-test] [!SUCCESS!]')
else:
printInfo('[py9kw-test] Test FAILED --> executing captcha_correct')
printInfo('[py9kw-test] Returned String: ' + result)
captchaSolver.setCaptchaCorrect(False)
printInfo('[py9kw-test] [!FAILURE!]')
creditsLeftNow = captchaSolver.getcredits()
# In selfsolve mode no credits will be used
if not selfsolve:
creditsUsed = creditsLeftBefore - creditsLeftNow
print(f'Credits used for this test: {creditsUsed}')
if creditsUsed > 0 and result is not None and not result_is_correct:
print(f'Your {creditsUsed} used credits should soon get refunded as the obtained result was wrong')
elif creditsUsed > 0 and result is None:
print('Your %d user credits should soon get refunded as you did not get any result' % creditsUsed)
elif creditsUsed > 0:
print('Your %d used credits will never come back :(' % creditsUsed)
print('Credits left: %d' % creditsLeftNow)
printInfo('[py9kw-test] [!END!]')
exit(0)