Added Singleton class to inherit as needed

This commit is contained in:
2023-03-27 20:07:17 -05:00
parent 4b69622cc6
commit ca61d5348f
46 changed files with 6455 additions and 97 deletions

View File

@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
# flake8: noqa
from .brotli import (
decompress, Decompressor, compress, BrotliEncoderMode, DEFAULT_MODE,
Compressor, MODE_GENERIC, MODE_TEXT, MODE_FONT, error, Error
)

View File

@@ -0,0 +1,466 @@
# -*- coding: utf-8 -*-
import math
import enum
from ._brotli import ffi, lib
class Error(Exception):
"""
Raised whenever an error is encountered with compressing or decompressing
data using brotlipy.
.. versionadded:: 0.5.1
"""
pass
#: An alias of :class:`Error <brotli.Error>` that exists for compatibility with
#: the original C brotli module.
#:
#: .. versionadded: 0.5.1
error = Error
class BrotliEncoderMode(enum.IntEnum):
"""
Compression modes for the Brotli encoder.
.. versionadded:: 0.5.0
"""
#: Default compression mode. The compressor does not know anything in
#: advance about the properties of the input.
GENERIC = lib.BROTLI_MODE_GENERIC
#: Compression mode for UTF-8 format text input.
TEXT = lib.BROTLI_MODE_TEXT
#: Compression mode used in WOFF 2.0
FONT = lib.BROTLI_MODE_FONT
# Define some names for compatibility with the C module.
#: The default compression mode for brotli.
DEFAULT_MODE = BrotliEncoderMode(lib.BROTLI_DEFAULT_MODE)
#: A compression mode where the compressor does not know anything in advance
#: about the properties of the input.
#:
#: .. note:: This name is defined for compatibility with the Brotli C
#: extension. If you're not interested in that compatibility, it is
#: recommended that you use :class:`BrotliEncoderMode
#: <brotli.BrotliEncoderMode>` instead.
#:
#: .. versionadded:: 0.5.0
MODE_GENERIC = BrotliEncoderMode.GENERIC
#: A compression mode for UTF-8 format text input.
#:
#: .. note:: This name is defined for compatibility with the Brotli C
#: extension. If you're not interested in that compatibility, it is
#: recommended that you use :class:`BrotliEncoderMode
#: <brotli.BrotliEncoderMode>` instead.
#:
#: .. versionadded:: 0.5.0
MODE_TEXT = BrotliEncoderMode.TEXT
#: The compression mode used in WOFF 2.0.
#:
#: .. note:: This name is defined for compatibility with the Brotli C
#: extension. If you're not interested in that compatibility, it is
#: recommended that you use :class:`BrotliEncoderMode
#: <brotli.BrotliEncoderMode>` instead.
#:
#: .. versionadded:: 0.5.0
MODE_FONT = BrotliEncoderMode.FONT
def decompress(data):
"""
Decompress a complete Brotli-compressed string.
:param data: A bytestring containing Brotli-compressed data.
"""
d = Decompressor()
data = d.decompress(data)
d.finish()
return data
def compress(data,
mode=DEFAULT_MODE,
quality=lib.BROTLI_DEFAULT_QUALITY,
lgwin=lib.BROTLI_DEFAULT_WINDOW,
lgblock=0,
dictionary=b''):
"""
Compress a string using Brotli.
.. versionchanged:: 0.5.0
Added ``mode``, ``quality``, `lgwin``, ``lgblock``, and ``dictionary``
parameters.
:param data: A bytestring containing the data to compress.
:type data: ``bytes``
:param mode: The encoder mode.
:type mode: :class:`BrotliEncoderMode` or ``int``
:param quality: Controls the compression-speed vs compression-density
tradeoffs. The higher the quality, the slower the compression. The
range of this value is 0 to 11.
:type quality: ``int``
:param lgwin: The base-2 logarithm of the sliding window size. The range of
this value is 10 to 24.
:type lgwin: ``int``
:param lgblock: The base-2 logarithm of the maximum input block size. The
range of this value is 16 to 24. If set to 0, the value will be set
based on ``quality``.
:type lgblock: ``int``
:param dictionary: A pre-set dictionary for LZ77. Please use this with
caution: if a dictionary is used for compression, the same dictionary
**must** be used for decompression!
:type dictionary: ``bytes``
:returns: The compressed bytestring.
:rtype: ``bytes``
"""
# This method uses private variables on the Compressor object, and
# generally does a whole lot of stuff that's not supported by the public
# API. The goal here is to minimise the number of allocations and copies
# we have to do. Users should prefer this method over the Compressor if
# they know they have single-shot data.
compressor = Compressor(
mode=mode,
quality=quality,
lgwin=lgwin,
lgblock=lgblock,
dictionary=dictionary
)
compressed_data = compressor._compress(data, lib.BROTLI_OPERATION_FINISH)
assert lib.BrotliEncoderIsFinished(compressor._encoder) == lib.BROTLI_TRUE
assert (
lib.BrotliEncoderHasMoreOutput(compressor._encoder) == lib.BROTLI_FALSE
)
return compressed_data
def _validate_mode(val):
"""
Validate that the mode is valid.
"""
try:
val = BrotliEncoderMode(val)
except ValueError:
raise Error("%s is not a valid encoder mode" % val)
def _validate_quality(val):
"""
Validate that the quality setting is valid.
"""
if not (0 <= val <= 11):
raise Error(
"%d is not a valid quality, must be between 0 and 11" % val
)
def _validate_lgwin(val):
"""
Validate that the lgwin setting is valid.
"""
if not (10 <= val <= 24):
raise Error("%d is not a valid lgwin, must be between 10 and 24" % val)
def _validate_lgblock(val):
"""
Validate that the lgblock setting is valid.
"""
if (val != 0) and not (16 <= val <= 24):
raise Error(
"%d is not a valid lgblock, must be either 0 or between 16 and 24"
% val
)
def _set_parameter(encoder, parameter, parameter_name, val):
"""
This helper function sets a specific Brotli encoder parameter, checking
the return code and raising :class:`Error <brotli.Error>` if it is
invalid.
"""
rc = lib.BrotliEncoderSetParameter(encoder, parameter, val)
if parameter == lib.BROTLI_PARAM_MODE:
_validate_mode(val)
elif parameter == lib.BROTLI_PARAM_QUALITY:
_validate_quality(val)
elif parameter == lib.BROTLI_PARAM_LGWIN:
_validate_lgwin(val)
elif parameter == lib.BROTLI_PARAM_LGBLOCK:
_validate_lgblock(val)
else: # pragma: no cover
raise RuntimeError("Unexpected parameter!")
# This block is defensive: I see no way to hit it, but as long as the
# function returns a value we can live in hope that the brotli folks will
# enforce their own constraints.
if rc != lib.BROTLI_TRUE: # pragma: no cover
raise Error(
"Error setting parameter %s: %d" % (parameter_name, val)
)
class Compressor(object):
"""
An object that allows for streaming compression of data using the Brotli
compression algorithm.
.. versionadded:: 0.5.0
:param mode: The encoder mode.
:type mode: :class:`BrotliEncoderMode` or ``int``
:param quality: Controls the compression-speed vs compression-density
tradeoffs. The higher the quality, the slower the compression. The
range of this value is 0 to 11.
:type quality: ``int``
:param lgwin: The base-2 logarithm of the sliding window size. The range of
this value is 10 to 24.
:type lgwin: ``int``
:param lgblock: The base-2 logarithm of the maximum input block size. The
range of this value is 16 to 24. If set to 0, the value will be set
based on ``quality``.
:type lgblock: ``int``
:param dictionary: A pre-set dictionary for LZ77. Please use this with
caution: if a dictionary is used for compression, the same dictionary
**must** be used for decompression!
:type dictionary: ``bytes``
"""
_dictionary = None
_dictionary_size = None
def __init__(self,
mode=DEFAULT_MODE,
quality=lib.BROTLI_DEFAULT_QUALITY,
lgwin=lib.BROTLI_DEFAULT_WINDOW,
lgblock=0,
dictionary=b''):
enc = lib.BrotliEncoderCreateInstance(
ffi.NULL, ffi.NULL, ffi.NULL
)
if not enc: # pragma: no cover
raise RuntimeError("Unable to allocate Brotli encoder!")
enc = ffi.gc(enc, lib.BrotliEncoderDestroyInstance)
# Configure the encoder appropriately.
_set_parameter(enc, lib.BROTLI_PARAM_MODE, "mode", mode)
_set_parameter(enc, lib.BROTLI_PARAM_QUALITY, "quality", quality)
_set_parameter(enc, lib.BROTLI_PARAM_LGWIN, "lgwin", lgwin)
_set_parameter(enc, lib.BROTLI_PARAM_LGBLOCK, "lgblock", lgblock)
if dictionary:
self._dictionary = ffi.new("uint8_t []", dictionary)
self._dictionary_size = len(dictionary)
lib.BrotliEncoderSetCustomDictionary(
enc, self._dictionary_size, self._dictionary
)
self._encoder = enc
def _compress(self, data, operation):
"""
This private method compresses some data in a given mode. This is used
because almost all of the code uses the exact same setup. It wouldn't
have to, but it doesn't hurt at all.
"""
# The 'algorithm' for working out how big to make this buffer is from
# the Brotli source code, brotlimodule.cc.
original_output_size = int(
math.ceil(len(data) + (len(data) >> 2) + 10240)
)
available_out = ffi.new("size_t *")
available_out[0] = original_output_size
output_buffer = ffi.new("uint8_t []", available_out[0])
ptr_to_output_buffer = ffi.new("uint8_t **", output_buffer)
input_size = ffi.new("size_t *", len(data))
input_buffer = ffi.new("uint8_t []", data)
ptr_to_input_buffer = ffi.new("uint8_t **", input_buffer)
rc = lib.BrotliEncoderCompressStream(
self._encoder,
operation,
input_size,
ptr_to_input_buffer,
available_out,
ptr_to_output_buffer,
ffi.NULL
)
if rc != lib.BROTLI_TRUE: # pragma: no cover
raise Error("Error encountered compressing data.")
assert not input_size[0]
size_of_output = original_output_size - available_out[0]
return ffi.buffer(output_buffer, size_of_output)[:]
def compress(self, data):
"""
Incrementally compress more data.
:param data: A bytestring containing data to compress.
:returns: A bytestring containing some compressed data. May return the
empty bytestring if not enough data has been inserted into the
compressor to create the output yet.
"""
return self._compress(data, lib.BROTLI_OPERATION_PROCESS)
def flush(self):
"""
Flush the compressor. This will emit the remaining output data, but
will not destroy the compressor. It can be used, for example, to ensure
that given chunks of content will decompress immediately.
"""
chunks = []
chunks.append(self._compress(b'', lib.BROTLI_OPERATION_FLUSH))
while lib.BrotliEncoderHasMoreOutput(self._encoder) == lib.BROTLI_TRUE:
chunks.append(self._compress(b'', lib.BROTLI_OPERATION_FLUSH))
return b''.join(chunks)
def finish(self):
"""
Finish the compressor. This will emit the remaining output data and
transition the compressor to a completed state. The compressor cannot
be used again after this point, and must be replaced.
"""
chunks = []
while lib.BrotliEncoderIsFinished(self._encoder) == lib.BROTLI_FALSE:
chunks.append(self._compress(b'', lib.BROTLI_OPERATION_FINISH))
return b''.join(chunks)
class Decompressor(object):
"""
An object that allows for streaming decompression of Brotli-compressed
data.
.. versionchanged:: 0.5.0
Added ``dictionary`` parameter.
:param dictionary: A pre-set dictionary for LZ77. Please use this with
caution: if a dictionary is used for compression, the same dictionary
**must** be used for decompression!
:type dictionary: ``bytes``
"""
_dictionary = None
_dictionary_size = None
def __init__(self, dictionary=b''):
dec = lib.BrotliDecoderCreateInstance(ffi.NULL, ffi.NULL, ffi.NULL)
self._decoder = ffi.gc(dec, lib.BrotliDecoderDestroyInstance)
if dictionary:
self._dictionary = ffi.new("uint8_t []", dictionary)
self._dictionary_size = len(dictionary)
lib.BrotliDecoderSetCustomDictionary(
self._decoder,
self._dictionary_size,
self._dictionary
)
def decompress(self, data):
"""
Decompress part of a complete Brotli-compressed string.
:param data: A bytestring containing Brotli-compressed data.
:returns: A bytestring containing the decompressed data.
"""
chunks = []
available_in = ffi.new("size_t *", len(data))
in_buffer = ffi.new("uint8_t[]", data)
next_in = ffi.new("uint8_t **", in_buffer)
while True:
# Allocate a buffer that's hopefully overlarge, but if it's not we
# don't mind: we'll spin around again.
buffer_size = 5 * len(data)
available_out = ffi.new("size_t *", buffer_size)
out_buffer = ffi.new("uint8_t[]", buffer_size)
next_out = ffi.new("uint8_t **", out_buffer)
rc = lib.BrotliDecoderDecompressStream(self._decoder,
available_in,
next_in,
available_out,
next_out,
ffi.NULL)
# First, check for errors.
if rc == lib.BROTLI_DECODER_RESULT_ERROR:
error_code = lib.BrotliDecoderGetErrorCode(self._decoder)
error_message = lib.BrotliDecoderErrorString(error_code)
raise Error(
"Decompression error: %s" % ffi.string(error_message)
)
# Next, copy the result out.
chunk = ffi.buffer(out_buffer, buffer_size - available_out[0])[:]
chunks.append(chunk)
if rc == lib.BROTLI_DECODER_RESULT_NEEDS_MORE_INPUT:
assert available_in[0] == 0
break
elif rc == lib.BROTLI_DECODER_RESULT_SUCCESS:
break
else:
# It's cool if we need more output, we just loop again.
assert rc == lib.BROTLI_DECODER_RESULT_NEEDS_MORE_OUTPUT
return b''.join(chunks)
def flush(self):
"""
Complete the decompression, return whatever data is remaining to be
decompressed.
.. deprecated:: 0.4.0
This method is no longer required, as decompress() will now
decompress eagerly.
:returns: A bytestring containing the remaining decompressed data.
"""
return b''
def finish(self):
"""
Finish the decompressor. As the decompressor decompresses eagerly, this
will never actually emit any data. However, it will potentially throw
errors if a truncated or damaged data stream has been used.
Note that, once this method is called, the decompressor is no longer
safe for further use and must be thrown away.
"""
assert (
lib.BrotliDecoderHasMoreOutput(self._decoder) == lib.BROTLI_FALSE
)
if lib.BrotliDecoderIsFinished(self._decoder) == lib.BROTLI_FALSE:
raise Error("Decompression error: incomplete compressed stream.")
return b''

View File

@@ -0,0 +1,224 @@
# -*- coding: utf-8 -*-
import sys
from cffi import FFI
ffi = FFI()
libraries = ['libbrotli']
if 'win32' not in str(sys.platform).lower():
libraries.append('stdc++')
ffi.set_source(
"_brotli",
"""#include <brotli/decode.h>
#include <brotli/encode.h>
""",
libraries=libraries,
include_dirs=["libbrotli", "libbrotli/include"]
)
ffi.cdef("""
/* common/types.h */
typedef bool BROTLI_BOOL;
#define BROTLI_TRUE ...
#define BROTLI_FALSE ...
/* dec/state.h */
/* Allocating function pointer. Function MUST return 0 in the case of
failure. Otherwise it MUST return a valid pointer to a memory region of
at least size length. Neither items nor size are allowed to be 0.
opaque argument is a pointer provided by client and could be used to
bind function to specific object (memory pool). */
typedef void* (*brotli_alloc_func)(void* opaque, size_t size);
/* Deallocating function pointer. Function SHOULD be no-op in the case the
address is 0. */
typedef void (*brotli_free_func)(void* opaque, void* address);
/* dec/decode.h */
typedef enum {
/* Decoding error, e.g. corrupt input or memory allocation problem */
BROTLI_DECODER_RESULT_ERROR = 0,
/* Decoding successfully completed */
BROTLI_DECODER_RESULT_SUCCESS = 1,
/* Partially done; should be called again with more input */
BROTLI_DECODER_RESULT_NEEDS_MORE_INPUT = 2,
/* Partially done; should be called again with more output */
BROTLI_DECODER_RESULT_NEEDS_MORE_OUTPUT = 3
} BrotliDecoderResult;
typedef enum {...} BrotliDecoderErrorCode;
typedef ... BrotliDecoderState;
/* Creates the instance of BrotliDecoderState and initializes it.
|alloc_func| and |free_func| MUST be both zero or both non-zero. In the
case they are both zero, default memory allocators are used. |opaque| is
passed to |alloc_func| and |free_func| when they are called. */
BrotliDecoderState* BrotliDecoderCreateInstance(brotli_alloc_func,
brotli_free_func,
void *);
/* Deinitializes and frees BrotliDecoderState instance. */
void BrotliDecoderDestroyInstance(BrotliDecoderState* state);
/* Decompresses the data. Supports partial input and output.
Must be called with an allocated input buffer in |*next_in| and an
allocated output buffer in |*next_out|. The values |*available_in| and
|*available_out| must specify the allocated size in |*next_in| and
|*next_out| respectively.
After each call, |*available_in| will be decremented by the amount of
input bytes consumed, and the |*next_in| pointer will be incremented by
that amount. Similarly, |*available_out| will be decremented by the
amount of output bytes written, and the |*next_out| pointer will be
incremented by that amount. |total_out|, if it is not a null-pointer,
will be set to the number of bytes decompressed since the last state
initialization.
Input is never overconsumed, so |next_in| and |available_in| could be
passed to the next consumer after decoding is complete. */
BrotliDecoderResult BrotliDecoderDecompressStream(BrotliDecoderState* s,
size_t* available_in,
const uint8_t** next_in,
size_t* available_out,
uint8_t** next_out,
size_t* total_out);
/* Fills the new state with a dictionary for LZ77, warming up the
ringbuffer, e.g. for custom static dictionaries for data formats.
Not to be confused with the built-in transformable dictionary of Brotli.
|size| should be less or equal to 2^24 (16MiB), otherwise the dictionary
will be ignored. The dictionary must exist in memory until decoding is
done and is owned by the caller. To use:
1) Allocate and initialize state with BrotliCreateInstance
2) Use BrotliSetCustomDictionary
3) Use BrotliDecompressStream
4) Clean up and free state with BrotliDestroyState
*/
void BrotliDecoderSetCustomDictionary(
BrotliDecoderState* s, size_t size, const uint8_t* dict);
/* Returns true, if decoder has some unconsumed output.
Otherwise returns false. */
BROTLI_BOOL BrotliDecoderHasMoreOutput(const BrotliDecoderState* s);
/* Returns true, if decoder has already received some input bytes.
Otherwise returns false. */
BROTLI_BOOL BrotliDecoderIsUsed(const BrotliDecoderState* s);
/* Returns true, if decoder is in a state where we reached the end of the
input and produced all of the output; returns false otherwise. */
BROTLI_BOOL BrotliDecoderIsFinished(const BrotliDecoderState* s);
/* Returns detailed error code after BrotliDecompressStream returns
BROTLI_DECODER_RESULT_ERROR. */
BrotliDecoderErrorCode BrotliDecoderGetErrorCode(
const BrotliDecoderState* s);
const char* BrotliDecoderErrorString(BrotliDecoderErrorCode c);
/* enc/encode.h */
typedef ... BrotliEncoderState;
typedef enum BrotliEncoderParameter {
BROTLI_PARAM_MODE = 0,
/* Controls the compression-speed vs compression-density tradeoffs. The
higher the quality, the slower the compression. Range is 0 to 11. */
BROTLI_PARAM_QUALITY = 1,
/* Base 2 logarithm of the sliding window size. Range is 10 to 24. */
BROTLI_PARAM_LGWIN = 2,
/* Base 2 logarithm of the maximum input block size. Range is 16 to 24.
If set to 0, the value will be set based on the quality. */
BROTLI_PARAM_LGBLOCK = 3
} BrotliEncoderParameter;
typedef enum BrotliEncoderMode {
/* Default compression mode. The compressor does not know anything in
advance about the properties of the input. */
BROTLI_MODE_GENERIC = 0,
/* Compression mode for UTF-8 format text input. */
BROTLI_MODE_TEXT = 1,
/* Compression mode used in WOFF 2.0. */
BROTLI_MODE_FONT = 2
} BrotliEncoderMode;
int BROTLI_DEFAULT_QUALITY = 11;
int BROTLI_DEFAULT_WINDOW = 22;
#define BROTLI_DEFAULT_MODE ...
typedef enum BrotliEncoderOperation {
BROTLI_OPERATION_PROCESS = 0,
/* Request output stream to flush. Performed when input stream is
depleted and there is enough space in output stream. */
BROTLI_OPERATION_FLUSH = 1,
/* Request output stream to finish. Performed when input stream is
depleted and there is enough space in output stream. */
BROTLI_OPERATION_FINISH = 2
} BrotliEncoderOperation;
/* Creates the instance of BrotliEncoderState and initializes it.
|alloc_func| and |free_func| MUST be both zero or both non-zero. In the
case they are both zero, default memory allocators are used. |opaque| is
passed to |alloc_func| and |free_func| when they are called. */
BrotliEncoderState* BrotliEncoderCreateInstance(brotli_alloc_func,
brotli_free_func,
void *);
/* Deinitializes and frees BrotliEncoderState instance. */
void BrotliEncoderDestroyInstance(BrotliEncoderState* state);
/* Compresses the data in |input_buffer| into |encoded_buffer|, and sets
|*encoded_size| to the compressed length.
BROTLI_DEFAULT_QUALITY, BROTLI_DEFAULT_WINDOW and BROTLI_DEFAULT_MODE
should be used as |quality|, |lgwin| and |mode| if there are no specific
requirements to encoder speed and compression ratio.
If compression fails, |*encoded_size| is set to 0.
If BrotliEncoderMaxCompressedSize(|input_size|) is not zero, then
|*encoded_size| is never set to the bigger value.
Returns false if there was an error and true otherwise. */
BROTLI_BOOL BrotliEncoderCompress(int quality,
int lgwin,
BrotliEncoderMode mode,
size_t input_size,
const uint8_t* input_buffer,
size_t* encoded_size,
uint8_t* encoded_buffer);
BROTLI_BOOL BrotliEncoderCompressStream(BrotliEncoderState* s,
BrotliEncoderOperation op,
size_t* available_in,
const uint8_t** next_in,
size_t* available_out,
uint8_t** next_out,
size_t* total_out);
BROTLI_BOOL BrotliEncoderSetParameter(BrotliEncoderState* state,
BrotliEncoderParameter p,
uint32_t value);
/* Fills the new state with a dictionary for LZ77, warming up the
ringbuffer, e.g. for custom static dictionaries for data formats.
Not to be confused with the built-in transformable dictionary of Brotli.
To decode, use BrotliSetCustomDictionary() of the decoder with the same
dictionary. */
void BrotliEncoderSetCustomDictionary(BrotliEncoderState* state,
size_t size,
const uint8_t* dict);
/* Check if encoder is in "finished" state, i.e. no more input is
acceptable and no more output will be produced.
Works only with BrotliEncoderCompressStream workflow.
Returns 1 if stream is finished and 0 otherwise. */
BROTLI_BOOL BrotliEncoderIsFinished(BrotliEncoderState* s);
/* Check if encoder has more output bytes in internal buffer.
Works only with BrotliEncoderCompressStream workflow.
Returns 1 if has more output (in internal buffer) and 0 otherwise. */
BROTLI_BOOL BrotliEncoderHasMoreOutput(BrotliEncoderState* s);
""")
if __name__ == '__main__':
ffi.compile()

View File

@@ -3,6 +3,8 @@ import os
import time
import threading
import requests
import json
from . import brotli
# Lib imports
import gi
@@ -141,25 +143,45 @@ class Plugin(PluginBase):
response = requests.post(tlink, headers=self._headers, data=from_translate)
if response.status_code == 200:
data = response.json()
self._translate_to_buffer.set_text(data["translated"])
data = self.get_data(response)
self.translate_tries = 0
self._translate_to_buffer.set_text(data["translated"])
if data["detected_language"]:
self._detected_language_lbl.set_label(f"Detected Language: {data['detected_language']}")
else:
self._detected_language_lbl.set_label(f"Selected Language: {self.from_trans}")
elif response.status_code >= 400 or response.status_code < 500:
self.get_vqd()
if not self.translate_tries > 4:
if not self.translate_tries > 2:
self._translate()
else:
msg = f"Could not translate... Response Code: {response.status_code}"
self._translate_to_buffer.set_text(msg)
def get_data(self, response):
data = None
try:
data = response.json()
except Exception as e:
...
try:
data = json.loads(response.text)
except Exception as e:
...
try:
decompress_str = brotli.decompress(response.content).decode("utf-8")
data = json.loads(decompress_str)
except Exception as e:
...
return data
# NOTE: https://github.com/deedy5/duckduckgo_search/blob/72acb900a346be576f0917dd3d6c0fbd618a71bf/duckduckgo_search/utils.py
def get_vqd(self):
response = requests.post(self.vqd_link, headers=self.vqd_headers, data=self.vqd_data, timeout=10)
response = requests.post(self.vqd_link, headers=self.vqd_headers, data=self.vqd_data, timeout=2)
if response.status_code == 200:
data = response.content
vqd_start_index = data.index(b"vqd='") + 5